From edc5ba748adf5223b8d23ab939e9c9b788e48ff7 Mon Sep 17 00:00:00 2001 From: Jared Ping <13906915+JaredP94@users.noreply.github.com> Date: Mon, 2 Nov 2026 23:18:46 +0200 Subject: [PATCH 1/5] fix(engine-manager): keep passive probes outside lifecycle locks Adapt NVIDIA/Personal-AI-Router#180 to share bounded status observations and discard them after lifecycle or health changes. Slow readiness no longer blocks start, stop, or port changes. Model sweeps include status and operation waits in their deadline and return HTTP 503 on timeout so peers retain their last-good lists. Preserve the fork's CLI detection, capability snapshots, and concurrent engine shutdown. The existing unreleased engine-manager 1.4.0 bump already covers this fix relative to v0.8.0. Co-authored-by: Kaylee Lubick Signed-off-by: Jared Ping <13906915+JaredP94@users.noreply.github.com> --- services/nvpair-engine-manager/README.md | 13 + services/nvpair-engine-manager/actions.go | 4 +- .../nvpair-engine-manager/desiredstate.go | 4 +- services/nvpair-engine-manager/executor.go | 33 +- services/nvpair-engine-manager/go.mod | 12 +- services/nvpair-engine-manager/go.sum | 10 + services/nvpair-engine-manager/httpserver.go | 35 +- .../nvpair-engine-manager/httpserver_test.go | 159 ++++++++ services/nvpair-engine-manager/install.go | 8 +- services/nvpair-engine-manager/lifecycle.go | 28 +- services/nvpair-engine-manager/models.go | 67 ++-- services/nvpair-engine-manager/setport.go | 4 +- services/nvpair-engine-manager/status.go | 270 +++++++++++-- .../status_models_test.go | 51 +++ services/nvpair-engine-manager/status_test.go | 356 ++++++++++++++++++ 15 files changed, 958 insertions(+), 96 deletions(-) create mode 100644 services/nvpair-engine-manager/httpserver_test.go create mode 100644 services/nvpair-engine-manager/status_models_test.go create mode 100644 services/nvpair-engine-manager/status_test.go diff --git a/services/nvpair-engine-manager/README.md b/services/nvpair-engine-manager/README.md index 6b380a5d1..d6b0afbe2 100644 --- a/services/nvpair-engine-manager/README.md +++ b/services/nvpair-engine-manager/README.md @@ -242,6 +242,19 @@ a successful command must leave the tool detectable. | `-32011` | `unsupported-platform`: the node is not `darwin/arm64`. | | `-32012` | `install-failed`: the install command failed or macmon was not detected afterwards. Command failures include only the last 20 stderr lines, capped to the last 4096 bytes; stdout is discarded. | +### Passive status and model deadlines + +Status and get-installed share bounded presence probes without holding the +lifecycle lock. A normal status read during an operation returns its current +snapshot; explicit port checks and model sweeps wait within their caller's +budget. Observations are discarded if a lifecycle operation or a newer state +change intervened, so a delayed readiness reply cannot undo stop or set-port. + +The five-second model sweep budget includes status probes and operation waits. +`GET /v1/models` returns HTTP 503 if the sweep cannot complete in time, rather +than a successful partial or empty list; peers retain their last-good inventory. +The local `engine:models` result remains best-effort with the same payload. + ### Effective capabilities `EngineStatus` = `{ engine, display_name, installed, running, healthy, port?, capabilities }`. diff --git a/services/nvpair-engine-manager/actions.go b/services/nvpair-engine-manager/actions.go index 1efe5d947..e61ab1d72 100644 --- a/services/nvpair-engine-manager/actions.go +++ b/services/nvpair-engine-manager/actions.go @@ -64,8 +64,8 @@ func (e *Executor) Action(ctx context.Context, engine, action string, params jso // model must not rewrite the engine's saved ON/OFF intent. doStop/doStart // are the intent-neutral pair (StopAll uses them for the same reason). func (e *Executor) restartAfterAction(ctx context.Context, st *engineState, engine, action string) error { - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() st.mu.Lock() running := st.running st.mu.Unlock() diff --git a/services/nvpair-engine-manager/desiredstate.go b/services/nvpair-engine-manager/desiredstate.go index 0c877dc10..238b48b55 100644 --- a/services/nvpair-engine-manager/desiredstate.go +++ b/services/nvpair-engine-manager/desiredstate.go @@ -139,8 +139,8 @@ func (e *Executor) restoreEnabled(ctx context.Context, engine string) error { if err != nil { return err } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() enabled, known, err := e.desired.get(engine) if err != nil || !known || !enabled { return err diff --git a/services/nvpair-engine-manager/executor.go b/services/nvpair-engine-manager/executor.go index 6f831b7d5..ef4c82037 100644 --- a/services/nvpair-engine-manager/executor.go +++ b/services/nvpair-engine-manager/executor.go @@ -47,14 +47,17 @@ type engineState struct { // none, in which case uninstall has nothing to preserve. modelsDir string - // opMu serializes lifecycle operations (install / start / stop / - // restart / uninstall) for this engine, so concurrent calls can't - // double-spawn or interleave. Lock order is opMu → mu; the per-op - // st.mu is released around probes, and the background health loop - // never takes opMu. + // opMu serializes lifecycle operations. Passive status probes never take + // it; lockOperation invalidates their observations before state can change. + // Lock order is opMu → mu; background health only takes mu. opMu sync.Mutex - mu sync.Mutex + mu sync.Mutex + // revision changes at both ends of a lifecycle operation and on passive + // state changes. A probe may only commit against its captured revision. + revision uint64 + opDone chan struct{} + statusRuns map[statusProbeKey]*statusProbeRun installed bool running bool healthy bool @@ -69,6 +72,24 @@ type engineState struct { startCancel context.CancelFunc } +// lockOperation marks the entire serialized operation, including external I/O. +func (st *engineState) lockOperation() { + st.opMu.Lock() + st.mu.Lock() + st.opDone = make(chan struct{}) + st.revision++ + st.mu.Unlock() +} + +func (st *engineState) unlockOperation() { + st.mu.Lock() + st.revision++ + close(st.opDone) + st.opDone = nil + st.mu.Unlock() + st.opMu.Unlock() +} + // Executor owns engine lifecycle for every engine known on this host. // Methods are synchronous and safe for concurrent use; the JSON-RPC // layer runs the long ones (install, start) in goroutines so the read diff --git a/services/nvpair-engine-manager/go.mod b/services/nvpair-engine-manager/go.mod index 390bd46ac..bda6a22a6 100644 --- a/services/nvpair-engine-manager/go.mod +++ b/services/nvpair-engine-manager/go.mod @@ -2,9 +2,17 @@ module nvpair-engine-manager go 1.25.0 -require nvpair-shared v0.0.0-00010101000000-000000000000 +require ( + github.com/stretchr/testify v1.11.1 + nvpair-shared v0.0.0-00010101000000-000000000000 +) -require github.com/google/uuid v1.6.0 // indirect +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) replace nvpair-shared => ../shared diff --git a/services/nvpair-engine-manager/go.sum b/services/nvpair-engine-manager/go.sum index 02c777231..502ed9d2d 100644 --- a/services/nvpair-engine-manager/go.sum +++ b/services/nvpair-engine-manager/go.sum @@ -1,6 +1,16 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/services/nvpair-engine-manager/httpserver.go b/services/nvpair-engine-manager/httpserver.go index 2c2eb6f06..9d8c6bcc0 100644 --- a/services/nvpair-engine-manager/httpserver.go +++ b/services/nvpair-engine-manager/httpserver.go @@ -93,18 +93,7 @@ func serveHTTP(ctx context.Context, port int, exec *Executor, mesh *clustertrust // The engine-native paths answer with PAIR's own type, mapped field by field // from the engine's admin API, so the engine's api_key has nowhere to go. func emSurface(exec *Executor, mesh *clustertrust.Mesh) (plainMux, tlsMux *http.ServeMux) { - models := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - return - } - w.Header().Set("Content-Type", "application/json") - // {"models":[...union...],"modelsByEngine":{"ollama":[...],...}, - // "loadedByEngine":{"ollama":[...],...}} — the flat union stays for - // existing consumers; modelsByEngine and loadedByEngine are additive - // (both omitempty, so absent when no engine reports them). - _ = json.NewEncoder(w).Encode(exec.ModelsResult(r.Context())) - }) + models := modelsHandler(exec) stats := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -136,6 +125,28 @@ func emSurface(exec *Executor, mesh *clustertrust.Mesh) (plainMux, tlsMux *http. return plainMux, tlsMux } +func modelsHandler(exec *Executor) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + result, err := exec.modelsResult(r.Context()) + if err != nil { + // The scanner keeps its last-good inventory on a non-200 reply. + // A timed-out sweep must not look like an authoritative empty list. + http.Error(w, "model inventory unavailable", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + // {"models":[...union...],"modelsByEngine":{"ollama":[...],...}, + // "loadedByEngine":{"ollama":[...],...}} — the flat union stays for + // existing consumers; modelsByEngine and loadedByEngine are additive + // (both omitempty, so absent when no engine reports them). + _ = json.NewEncoder(w).Encode(result) + } +} + // serveOrWarn runs srv on l, treating a normal shutdown (and the sub-listener // closing under it) as success. func serveOrWarn(srv *http.Server, l net.Listener) { diff --git a/services/nvpair-engine-manager/httpserver_test.go b/services/nvpair-engine-manager/httpserver_test.go new file mode 100644 index 000000000..201021827 --- /dev/null +++ b/services/nvpair-engine-manager/httpserver_test.go @@ -0,0 +1,159 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" +) + +type healthProbeTransport func(*http.Request) (*http.Response, error) + +func (f healthProbeTransport) RoundTrip(r *http.Request) (*http.Response, error) { + return f(r) +} + +// Inventory must report a timeout as unavailable. A successful empty list +// would make clients discard models when only the status probe was slow. +func TestModelsHandler_RejectsSweepThatExpiresDuringStatusProbe(t *testing.T) { + fixture := newBlockedReadinessFixture(t) + ex := fixture.executor + enableModelInventory(t, ex) + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + req := httptest.NewRequest(http.MethodGet, modelsPath, nil).WithContext(ctx) + response := httptest.NewRecorder() + // Signal that the handler returned so its deadline can be checked separately. + finished := make(chan struct{}) + go func() { + modelsHandler(ex)(response, req) + close(finished) + }() + awaitProbeStart(t, fixture.probeStarted) + select { + case <-finished: + case <-time.After(time.Second): + fixture.releaseProbe() + require.FailNow(t, "inventory response ignored its deadline") + } + require.Equal(t, http.StatusServiceUnavailable, response.Code) + fixture.releaseProbe() +} + +// Model requests must stop waiting for an engine operation at the caller's +// deadline or the service's own limit, whichever expires first. +func TestModelsHandler_EnforcesDeadlinesDuringEngineOperation(t *testing.T) { + test := func(name string, callerTimeout, wantElapsed time.Duration) { + t.Run(name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m := testEngineManifest(filepath.Join(t.TempDir(), "missing"+exeExt())) + platform := m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] + platform.Runtime.Ready = nil + platform.Runtime.Health = nil + m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] = platform + ex := newTestExecutor(t, m) + enableModelInventory(t, ex) + st, err := ex.state("fake") + require.NoError(t, err) + st.lockOperation() + defer st.unlockOperation() + ctx := context.Background() + if callerTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, callerTimeout) + defer cancel() + } + req := httptest.NewRequest(http.MethodGet, modelsPath, nil).WithContext(ctx) + response := httptest.NewRecorder() + // Observe completion while the operation stays held. The outer + // limit catches a handler that ignores its deadline. + finished := make(chan struct{}) + started := time.Now() + go func() { + modelsHandler(ex)(response, req) + close(finished) + }() + select { + case <-finished: + case <-time.After(wantElapsed + 2*time.Second): + t.Fatal("model request kept waiting past its deadline") + } + require.Equal(t, wantElapsed, time.Since(started)) + require.Equal(t, http.StatusServiceUnavailable, response.Code) + }) + }) + } + test("honors shorter caller deadline", 50*time.Millisecond, 50*time.Millisecond) + test("enforces service deadline without caller deadline", 0, 5*time.Second) +} + +// Keep the operation active until synctest confirms the model request is +// waiting. An in-memory engine response avoids network scheduling in the test. +func TestModelsHandler_ReturnsModelsAfterEngineOperation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m := testEngineManifest(filepath.Join(t.TempDir(), "missing"+exeExt())) + platform := m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] + platform.Runtime.Ready = nil + platform.Runtime.Health = nil + m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] = platform + ex := newTestExecutor(t, m) + enableModelInventory(t, ex) + ex.client = &http.Client{Transport: healthProbeTransport(func(req *http.Request) (*http.Response, error) { + if req.Method != http.MethodGet || req.URL.Path != "/api/tags" { + return nil, fmt.Errorf("unexpected model request: %s %s", req.Method, req.URL.Path) + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"models":[{"name":"fixture-model"}]}`)), + }, nil + })} + st, err := ex.state("fake") + require.NoError(t, err) + st.mu.Lock() + st.running = true + st.healthy = true + st.adopted = true + st.mu.Unlock() + st.lockOperation() + var once sync.Once + unlock := func() { once.Do(st.unlockOperation) } + defer unlock() + req := httptest.NewRequest(http.MethodGet, modelsPath, nil) + response := httptest.NewRecorder() + // Distinguish a request waiting for the operation from an early reply. + finished := make(chan struct{}) + go func() { + modelsHandler(ex)(response, req) + close(finished) + }() + synctest.Wait() + select { + case <-finished: + t.Fatal("model request returned before the operation finished") + default: + } + unlock() + awaitResult(t, finished) + require.Equal(t, http.StatusOK, response.Code) + require.Equal(t, "application/json", response.Header().Get("Content-Type")) + var result ModelsResult + require.NoError(t, json.Unmarshal(response.Body.Bytes(), &result)) + require.Equal(t, []string{"fixture-model"}, result.Models) + require.Equal(t, map[string][]string{"fake": {"fixture-model"}}, result.ByEngine) + }) +} diff --git a/services/nvpair-engine-manager/install.go b/services/nvpair-engine-manager/install.go index 1efd785d8..bcba61def 100644 --- a/services/nvpair-engine-manager/install.go +++ b/services/nvpair-engine-manager/install.go @@ -27,8 +27,8 @@ func (e *Executor) Install(ctx context.Context, engine string) error { if err != nil { return err } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() if ok, _ := e.Detect(engine); ok { e.reporter.clear(installFailedID(engine)) e.emitInstallProgress(engine, "already-installed", 100) @@ -184,8 +184,8 @@ func (e *Executor) Uninstall(ctx context.Context, engine string) error { if err != nil { return err } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() if ok, _ := e.Detect(engine); !ok { // Already gone — by its own uninstaller, or by hand. Drop our ownership // claim with it: left behind, it would authorise removing a copy the diff --git a/services/nvpair-engine-manager/lifecycle.go b/services/nvpair-engine-manager/lifecycle.go index ed2b48fa5..a36369da5 100644 --- a/services/nvpair-engine-manager/lifecycle.go +++ b/services/nvpair-engine-manager/lifecycle.go @@ -90,8 +90,8 @@ func (e *Executor) StartWith(ctx context.Context, engine string, opts startOpts) if err != nil { return err } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() if err := e.doStart(ctx, st, engine, opts); err != nil { return err } @@ -334,6 +334,7 @@ func (e *Executor) watch(st *engineState, engine string, proc *managedProc) { st.proc = nil st.running = false st.healthy = false + st.revision++ if st.healthStop != nil { st.healthStop() st.healthStop = nil @@ -358,8 +359,8 @@ func (e *Executor) Stop(engine string) error { if err != nil { return err } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() stopErr := e.doStop(st, engine) // A user-initiated Stop is explicit OFF intent. Persist it even when // doStop declines a genuinely-foreign listener we can't terminate, so the @@ -376,6 +377,8 @@ func (e *Executor) doStop(st *engineState, engine string) error { mode := rt.modeOrDefault() st.mu.Lock() if !st.running { + // Invalidate passive adoption and preserve explicit OFF intent. + st.stopping = true st.mu.Unlock() return nil } @@ -541,7 +544,11 @@ func (e *Executor) markStopped(st *engineState, engine string) { // live listener returning 503 is unhealthy, not stopped. A successful control // command is likewise not proof that its daemon actually exited. func (e *Executor) waitUnavailable(p *Probe, port int, timeout time.Duration) bool { - ctx, cancel := context.WithTimeout(context.Background(), timeout) + return e.waitUnavailableContext(context.Background(), p, port, timeout) +} + +func (e *Executor) waitUnavailableContext(parent context.Context, p *Probe, port int, timeout time.Duration) bool { + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() misses := 0 for { @@ -681,8 +688,8 @@ func (e *Executor) Restart(ctx context.Context, engine string) error { if err != nil { return err } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() if err := e.doStop(st, engine); err != nil { return err } @@ -724,8 +731,8 @@ func (e *Executor) StopAll() { if cancel != nil { cancel() } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() if err := e.doStop(st, n); err != nil { slog.Warn("engine stop during shutdown failed", "engine", n, "err", err) } @@ -757,6 +764,9 @@ func (e *Executor) runHealth(ctx context.Context, st *engineState, engine string } was := st.healthy st.healthy = ok + if was != ok { + st.revision++ + } st.mu.Unlock() switch { case was && !ok: diff --git a/services/nvpair-engine-manager/models.go b/services/nvpair-engine-manager/models.go index 9714902a7..171b2ed50 100644 --- a/services/nvpair-engine-manager/models.go +++ b/services/nvpair-engine-manager/models.go @@ -61,24 +61,31 @@ func (e *Executor) Models(ctx context.Context) []string { // the server is down. Best-effort: a per-engine failure is logged and skipped // so the engines that did answer still surface their models. // -// The per-engine queries run CONCURRENTLY. A host running several engines (e.g. -// Ollama + LM Studio) must not have the later engines squeezed out of a single -// shared deadline: sequential queries under one budget meant a slow first engine -// left little for the second, and the peer's discovery daemon fetches this over -// HTTP with its own (shorter) timeout, so a squeezed second engine silently -// dropped its models for peers. Concurrency bounds the sweep by the slowest -// single engine, not the sum. +// Queries run concurrently so a slow engine does not consume the time available +// to the others. The five-second limit covers status checks and model-list +// requests for every engine. func (e *Executor) ModelsResult(ctx context.Context) ModelsResult { + res, _ := e.modelsResult(ctx) + return res +} + +// modelsResult also reports a canceled sweep to HTTP callers, which must not +// replace a peer's last-good inventory with a successful partial response. +func (e *Executor) modelsResult(ctx context.Context) (ModelsResult, error) { ctx, cancel := context.WithTimeout(ctx, modelsTimeout) defer cancel() + type engineResult struct { + models []string + modelsOK bool + loaded []string + loadedOK bool + statusErr error + } engineNames := e.reg.Names() // Results indexed by engine position so the flattened output stays // deterministic (registration order) despite concurrent completion. - perEngine := make([][]string, len(engineNames)) - listOK := make([]bool, len(engineNames)) - loaded := make([][]string, len(engineNames)) - loadedOK := make([]bool, len(engineNames)) + results := make([]engineResult, len(engineNames)) var wg sync.WaitGroup for i, name := range engineNames { mf, ok := e.reg.Get(name) @@ -98,16 +105,21 @@ func (e *Executor) ModelsResult(ctx context.Context) ModelsResult { wg.Add(1) go func(i int, name string, listSpec, loadedSpec *ActionResult) { defer wg.Done() - st, err := e.Status(name) - if err != nil || !st.Running { + result := &results[i] + st, err := e.statusWithContext(ctx, name, 0, true) + if err != nil { + result.statusErr = err + return + } + if !st.Running { return } if listSpec != nil { if raw, err := e.Action(ctx, name, "list_models", nil); err != nil { slog.Debug("engine:models list_models failed", "engine", name, "err", err) } else if models, ok := extractStringsResult(raw, listSpec); ok { - perEngine[i] = models - listOK[i] = true + result.models = models + result.modelsOK = true } else { slog.Debug("engine:models list_models returned an invalid inventory", "engine", name) } @@ -118,9 +130,9 @@ func (e *Executor) ModelsResult(ctx context.Context) ModelsResult { } else if models, ok := extractStringsResult(raw, loadedSpec); ok { // A successful query — even an empty result — records the key // so a consumer can tell "running, nothing loaded" apart from - // "unknown". An action or shape error leaves loadedOK[i] false. - loaded[i] = models - loadedOK[i] = true + // "unknown". An action or shape error leaves loadedOK false. + result.loaded = models + result.loadedOK = true } else { slog.Debug("engine:models loaded_models returned an invalid inventory", "engine", name) } @@ -131,13 +143,14 @@ func (e *Executor) ModelsResult(ctx context.Context) ModelsResult { res := ModelsResult{Models: []string{}} seen := map[string]bool{} - for i, models := range perEngine { - if listOK[i] { + for i, result := range results { + if result.modelsOK { if res.ByEngine == nil { res.ByEngine = make(map[string][]string) } // Normalize nil -> [] so the wire distinguishes a successful empty // inventory from a missing/not-queryable engine. + models := result.models if models == nil { models = []string{} } @@ -149,20 +162,28 @@ func (e *Executor) ModelsResult(ctx context.Context) ModelsResult { } } } - if loadedOK[i] { + if result.loadedOK { if res.LoadedByEngine == nil { res.LoadedByEngine = make(map[string][]string) } // Normalize nil -> [] so the wire carries an empty array ("queried, // nothing resident"), never null. - ld := loaded[i] + ld := result.loaded if ld == nil { ld = []string{} } res.LoadedByEngine[engineNames[i]] = ld } } - return res + if err := ctx.Err(); err != nil { + return res, err + } + for _, result := range results { + if result.statusErr != nil { + return res, result.statusErr + } + } + return res, nil } // extractStrings is the value-only wrapper used by focused extractor tests. diff --git a/services/nvpair-engine-manager/setport.go b/services/nvpair-engine-manager/setport.go index 604cfea82..f1ae6de50 100644 --- a/services/nvpair-engine-manager/setport.go +++ b/services/nvpair-engine-manager/setport.go @@ -38,8 +38,8 @@ func (e *Executor) SetPort(ctx context.Context, engine string, port int) (Engine if err != nil { return EngineStatus{}, err } - st.opMu.Lock() - defer st.opMu.Unlock() + st.lockOperation() + defer st.unlockOperation() st.mu.Lock() wasRunning := st.running diff --git a/services/nvpair-engine-manager/status.go b/services/nvpair-engine-manager/status.go index b22ef240f..affe85b45 100644 --- a/services/nvpair-engine-manager/status.go +++ b/services/nvpair-engine-manager/status.go @@ -11,6 +11,24 @@ import ( const presenceRefusalWindow = 500 * time.Millisecond +// A shared passive probe has its own bound, so one caller's cancellation does +// not stop other readers. It is canceled early when its last reader leaves. +const passiveProbeTimeout = 4 * time.Second + +type statusProbeKey struct { + revision uint64 + port int +} + +type statusProbeRun struct { + done chan struct{} + ctx context.Context + cancel context.CancelFunc + waiters int + err error + stale bool +} + // presenceResult distinguishes a positively identified engine from an // occupied/indeterminate target. Callers that might spawn or install must fail // closed when Occupied is true: a listener that did not pass the identity probe @@ -21,25 +39,35 @@ type presenceResult struct { } // Detect updates and returns whether the engine is installed, based on -// the manifest's executable candidates (env-expanded existence checks). -// Command mode detects only the CLI that lifecycle/actions will execute. +// the manifest's detect paths (env-expanded existence checks). func (e *Executor) Detect(engine string) (bool, error) { st, err := e.state(engine) if err != nil { return false, err } - binPath := detectedExecutable(st.plat, st.installDir, runtime.GOOS) - installed := binPath != "" + installed, binPath := detectPath(st) st.mu.Lock() + changed := st.installed != installed || (binPath != "" && st.binPath != binPath) st.installed = installed if binPath != "" { st.binPath = binPath } + if changed { + st.revision++ + } st.mu.Unlock() return installed, nil } -// Status returns a fresh snapshot (re-detecting installed-ness). +// detectPath observes the filesystem without publishing a partial state. A +// passive status check applies this result together with its port observation. +func detectPath(st *engineState) (bool, string) { + binPath := detectedExecutable(st.plat, st.installDir, runtime.GOOS) + return binPath != "", binPath +} + +// Status refreshes installation and fixed-port presence when no lifecycle +// operation is active. During an operation it returns the current snapshot. func (e *Executor) Status(engine string) (EngineStatus, error) { return e.StatusAtPort(engine, 0) } @@ -48,6 +76,12 @@ func (e *Executor) Status(engine string) (EngineStatus, error) { // lets upgrade orchestration identify a legacy listener before changing the // persisted backend port; successful identification updates the live state. func (e *Executor) StatusAtPort(engine string, probePort int) (EngineStatus, error) { + return e.statusWithContext(context.Background(), engine, probePort, probePort > 0) +} + +// statusWithContext lets deadline-bound callers leave a shared presence probe +// while other status readers continue to use its result. +func (e *Executor) statusWithContext(ctx context.Context, engine string, probePort int, waitForOperation bool) (EngineStatus, error) { st, err := e.state(engine) if err != nil { // Known engine with no host-platform block: return a shell status @@ -58,42 +92,89 @@ func (e *Executor) StatusAtPort(engine string, probePort int) (EngineStatus, err } return EngineStatus{}, err } - st.opMu.Lock() - defer st.opMu.Unlock() - pathInstalled, _ := e.Detect(engine) - st.mu.Lock() - port := st.port - st.mu.Unlock() - if probePort > 0 { - port = probePort + for { + st.mu.Lock() + port := st.port + if probePort > 0 { + port = probePort + } + // Normal status reads can return the current transition state. A + // one-shot port probe or model sweep must wait for an authoritative + // post-operation observation, within its caller's deadline. + if st.opDone != nil { + if waitForOperation { + done := st.opDone + st.mu.Unlock() + select { + case <-done: + continue + case <-ctx.Done(): + return EngineStatus{}, ctx.Err() + } + } + result := statusLocked(engine, st) + st.mu.Unlock() + return result, nil + } + key := statusProbeKey{revision: st.revision, port: port} + if st.statusRuns == nil { + st.statusRuns = make(map[statusProbeKey]*statusProbeRun) + } + run := st.statusRuns[key] + if run == nil { + probeCtx, cancel := context.WithTimeout(context.Background(), passiveProbeTimeout) + run = &statusProbeRun{done: make(chan struct{}), ctx: probeCtx, cancel: cancel} + st.statusRuns[key] = run + go e.refreshPassiveStatus(engine, st, key, run) + } + run.waiters++ + st.mu.Unlock() + var waitErr error + select { + case <-run.done: + case <-ctx.Done(): + waitErr = ctx.Err() + } + cancelRun := false + st.mu.Lock() + run.waiters-- + if run.waiters == 0 && st.statusRuns[key] == run { + delete(st.statusRuns, key) + cancelRun = true + } + st.mu.Unlock() + if cancelRun { + run.cancel() + } + if waitErr != nil { + return EngineStatus{}, waitErr + } + if waitForOperation { + if run.err != nil { + return EngineStatus{}, run.err + } + if run.stale { + continue + } + st.mu.Lock() + if st.opDone != nil { + st.mu.Unlock() + continue + } + result := statusLocked(engine, st) + st.mu.Unlock() + return result, nil + } + return e.snapshot(engine, st), nil } - e.reconcilePresence(context.Background(), engine, st, pathInstalled, port, false) - return e.snapshot(engine, st), nil } // GetInstalled lists every known engine with its current status. func (e *Executor) GetInstalled() []EngineStatus { var out []EngineStatus for _, name := range e.reg.Names() { - st, err := e.state(name) - if err != nil { - // Known engine with no block for this host: surface a shell - // status so the UI can still list it as unavailable here. - dn := name - if m, ok := e.reg.Get(name); ok { - dn = m.DisplayName - } - out = append(out, EngineStatus{Engine: name, DisplayName: dn}) - continue - } - st.opMu.Lock() - pathInstalled, _ := e.Detect(name) - st.mu.Lock() - port := st.port - st.mu.Unlock() - e.reconcilePresence(context.Background(), name, st, pathInstalled, port, false) - out = append(out, e.snapshot(name, st)) - st.opMu.Unlock() + status, _ := e.Status(name) + out = append(out, status) } return out } @@ -115,6 +196,10 @@ func (e *Executor) Errors() []serviceError { func (e *Executor) snapshot(engine string, st *engineState) EngineStatus { st.mu.Lock() defer st.mu.Unlock() + return statusLocked(engine, st) +} + +func statusLocked(engine string, st *engineState) EngineStatus { return EngineStatus{ Engine: engine, DisplayName: st.manifest.DisplayName, @@ -126,6 +211,123 @@ func (e *Executor) snapshot(engine string, st *engineState) EngineStatus { } } +// refreshPassiveStatus stages filesystem and network observations without an +// engine operation lock. Its short commit discards the result if any lifecycle +// operation or another state update ran in the meantime. Concurrent readers of +// the same port and revision wait on one probe. +func (e *Executor) refreshPassiveStatus(engine string, st *engineState, key statusProbeKey, run *statusProbeRun) { + defer func() { + st.mu.Lock() + if st.statusRuns[key] == run { + delete(st.statusRuns, key) + } + close(run.done) + st.mu.Unlock() + }() + defer run.cancel() + ctx := run.ctx + + st.mu.Lock() + if st.opDone != nil || st.revision != key.revision { + run.stale = true + st.mu.Unlock() + return + } + running, adopted, proc := st.running, st.adopted, st.proc + ready := st.plat.Runtime.Ready + mode := st.plat.Runtime.modeOrDefault() + st.mu.Unlock() + + pathInstalled, binPath := detectPath(st) + checkPresence := pathInstalled || mode == "process" + checkPresence = checkPresence && key.port != 0 && ready != nil && proc == nil && (!running || adopted) + var listener listenerProbeResult + var identified, refusedConfirmed bool + if checkPresence { + probeCtx, stopProbe := context.WithTimeout(ctx, presenceRefusalWindow) + listener = probeListener(probeCtx, ready, key.port) + stopProbe() + switch listener { + case listenerProbeRefused: + if running && adopted { + refusedConfirmed = e.waitUnavailableContext(ctx, ready, key.port, presenceRefusalWindow) + } + case listenerProbeReachable: + identified = e.probe(ctx, ready, key.port) + } + } + if ctx.Err() != nil { + run.err = ctx.Err() + return + } + + st.mu.Lock() + if st.opDone != nil || st.revision != key.revision { + run.stale = true + st.mu.Unlock() + return + } + before := statusLocked(engine, st) + oldBinPath, oldAdopted := st.binPath, st.adopted + emit := false + st.installed = pathInstalled + if pathInstalled { + st.binPath = binPath + } + if checkPresence { + switch { + case listener == listenerProbeRefused && refusedConfirmed: + emit = before.Running || before.Healthy || oldAdopted || before.Installed != pathInstalled + if st.healthStop != nil { + st.healthStop() + st.healthStop = nil + } + st.running = false + st.healthy = false + st.adopted = false + if !pathInstalled { + st.binPath = "" + } + case listener == listenerProbeRefused || listener == listenerProbeIndeterminate: + // A single refusal or an uncertain transport result cannot prove + // an adopted service stopped. + if running && adopted { + st.installed = true + if !pathInstalled { + st.binPath = "" + } + } + case identified && !st.stopping: + emit = !before.Running || !before.Healthy || !oldAdopted || before.Port != key.port + st.installed = true + st.running = true + st.healthy = true + st.adopted = true + st.port = key.port + if !pathInstalled { + st.binPath = "" + } + case !identified && listener == listenerProbeReachable && running && adopted: + emit = before.Healthy + // A live listener that no longer identifies as this engine is + // unhealthy, but its port must not be treated as free. + st.installed = true + st.healthy = false + if !pathInstalled { + st.binPath = "" + } + } + } + after := statusLocked(engine, st) + if before != after || oldBinPath != st.binPath || oldAdopted != st.adopted { + st.revision++ + } + st.mu.Unlock() + if emit { + e.emitState(engine) + } +} + // reconcilePresence reconciles filesystem detection with a fixed-port engine // service. A successful engine-specific readiness probe is authoritative even // when the binary lives outside NVPAIR's managed paths. Conversely, readiness diff --git a/services/nvpair-engine-manager/status_models_test.go b/services/nvpair-engine-manager/status_models_test.go new file mode 100644 index 000000000..6f3179a42 --- /dev/null +++ b/services/nvpair-engine-manager/status_models_test.go @@ -0,0 +1,51 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "net/http" + + "testing" + + "github.com/stretchr/testify/require" +) + +// Canceling inventory should release its caller promptly. A later status read +// must still be able to identify the listener after that cancellation. +func TestModelsResult_CanceledCallerReturnsBeforeProbeCompletes(t *testing.T) { + fixture := newBlockedReadinessFixture(t) + ex := fixture.executor + enableModelInventory(t, ex) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // Receive cancellation while the fixture still withholds the engine reply. + finished := make(chan error, 1) + go func() { + _, err := ex.modelsResult(ctx) + finished <- err + }() + awaitProbeStart(t, fixture.probeStarted) + cancel() + require.ErrorIs(t, awaitResult(t, finished), context.Canceled) + // The last caller leaving must stop the HTTP request before its reply is + // released, rather than merely letting the caller stop waiting. + awaitResult(t, fixture.probeCanceled) + fixture.releaseProbe() + status, err := ex.Status("fake") + require.NoError(t, err) + require.True(t, status.Running) +} + +// enableModelInventory adds the result specification needed for modelsResult +// to include this engine in an inventory sweep. +func enableModelInventory(t *testing.T, ex *Executor) { + t.Helper() + m, ok := ex.reg.Get("fake") + require.True(t, ok) + m.Actions["list_models"] = Action{ + HTTP: &ActionHTTP{Method: http.MethodGet, Path: "/api/tags"}, + Result: &ActionResult{Array: "models", Field: "name"}, + } +} diff --git a/services/nvpair-engine-manager/status_test.go b/services/nvpair-engine-manager/status_test.go new file mode 100644 index 000000000..748f3cb24 --- /dev/null +++ b/services/nvpair-engine-manager/status_test.go @@ -0,0 +1,356 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "runtime" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" +) + +// Status and GetInstalled serve UI and discovery reads while lifecycle actions +// may spend seconds in external I/O. Neither read should wait for that action. +func TestStatusReads_ReturnDuringLifecycleOperation(t *testing.T) { + test := func(name string, read func(*Executor) (EngineStatus, error)) { + t.Run(name, func(t *testing.T) { + ex := newTestExecutor(t, testEngineManifest(filepath.Join(t.TempDir(), "missing"+exeExt()))) + st, err := ex.state("fake") + require.NoError(t, err) + st.lockOperation() + defer st.unlockOperation() + + // Run the read separately so a blocked call cannot hang the test. + // The buffer lets it finish even if the timeout fails the test. + finished := make(chan statusResult, 1) + go func() { + status, err := read(ex) + finished <- statusResult{status, err} + }() + select { + case got := <-finished: + require.NoError(t, got.err) + require.False(t, got.status.Running) + case <-time.After(time.Second): + require.FailNow(t, "status read waited for the lifecycle operation lock") + } + }) + } + test("status", func(ex *Executor) (EngineStatus, error) { return ex.Status("fake") }) + test("get installed", func(ex *Executor) (EngineStatus, error) { + statuses := ex.GetInstalled() + if len(statuses) != 1 { + return EngineStatus{}, fmt.Errorf("expected one engine, got %d", len(statuses)) + } + return statuses[0], nil + }) +} + +// A readiness reply can arrive after a user has stopped the engine. That old +// reply must not undo the explicit OFF intent by adopting the listener again. +func TestStatus_DoesNotReadoptAfterStopDuringProbe(t *testing.T) { + fixture := newBlockedReadinessFixture(t) + ex := fixture.executor + // Save the delayed status result so it can be checked after Stop. + finished := make(chan statusResult, 1) + go func() { + status, err := ex.Status("fake") + finished <- statusResult{status, err} + }() + awaitProbeStart(t, fixture.probeStarted) + + // Bound the wait for Stop so a lock regression fails promptly. + stopped := make(chan error, 1) + go func() { stopped <- ex.Stop("fake") }() + select { + case err := <-stopped: + require.NoError(t, err) + case <-time.After(time.Second): + fixture.releaseProbe() + require.FailNow(t, "Stop waited for a passive readiness probe") + } + fixture.releaseProbe() + got := awaitResult(t, finished) + require.NoError(t, got.err) + require.False(t, got.status.Running, "an old probe re-adopted after Stop") + status, err := ex.Status("fake") + require.NoError(t, err) + require.False(t, status.Running, "later status undid an explicit Stop") +} + +// A failed attempt to stop a managed orphan can leave it running with stopping +// still set. A successful status check must not turn that into a health failure. +func TestStatus_DoesNotMarkReadyStoppingEngineUnhealthy(t *testing.T) { + fixture := newBlockedReadinessFixture(t) + fixture.releaseProbe() + st, err := fixture.executor.state("fake") + require.NoError(t, err) + // Reproduce the state left by a failed stop without terminating a process. + st.mu.Lock() + st.running = true + st.healthy = true + st.adopted = true + st.stopping = true + st.mu.Unlock() + + status, err := fixture.executor.Status("fake") + require.NoError(t, err) + require.True(t, status.Running) + require.True(t, status.Healthy, "a successfully identified listener was marked unhealthy") +} + +// SetPort may finish while a status check is probing the old listener. Its +// late result must not restore the port that SetPort just replaced. +func TestStatus_DoesNotRestoreOldPortAfterSetPort(t *testing.T) { + fixture := newBlockedReadinessFixture(t) + ex := fixture.executor + ex.overrideDir = t.TempDir() + newPort, err := freePort() + require.NoError(t, err) + for newPort == fixture.port { + newPort, err = freePort() + require.NoError(t, err) + } + // Save the old-port status result for comparison after SetPort finishes. + finished := make(chan statusResult, 1) + go func() { + status, err := ex.Status("fake") + finished <- statusResult{status, err} + }() + awaitProbeStart(t, fixture.probeStarted) + + // Check that SetPort returns before allowing the old status reply through. + changed := make(chan statusResult, 1) + go func() { + status, err := ex.SetPort(context.Background(), "fake", newPort) + changed <- statusResult{status, err} + }() + select { + case got := <-changed: + require.NoError(t, got.err) + require.Equal(t, newPort, got.status.Port) + case <-time.After(time.Second): + fixture.releaseProbe() + require.FailNow(t, "SetPort waited for a passive readiness probe") + } + fixture.releaseProbe() + got := awaitResult(t, finished) + require.NoError(t, got.err) + require.Equal(t, newPort, got.status.Port, "an old probe restored the former port") + status, err := ex.Status("fake") + require.NoError(t, err) + require.Equal(t, newPort, status.Port) +} + +// Canceling one status reader must leave the shared readiness probe alive for +// another reader, so the remaining caller still receives the probe result. +func TestStatus_SecondReaderUsesProbeAfterFirstCancels(t *testing.T) { + fixture := newBlockedReadinessFixture(t) + ex := fixture.executor + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // Capture this caller's cancellation separately from the remaining reader. + first := make(chan error, 1) + go func() { + _, err := ex.statusWithContext(ctx, "fake", 0, false) + first <- err + }() + awaitProbeStart(t, fixture.probeStarted) + // Save the other caller's result to prove that its request survives. + second := make(chan statusResult, 1) + go func() { + status, err := ex.Status("fake") + second <- statusResult{status, err} + }() + state, err := ex.state("fake") + require.NoError(t, err) + // Wait until the second reader joins before canceling the first. Without + // this check, scheduling could make the test cancel the probe too early. + joined := false + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + state.mu.Lock() + for _, run := range state.statusRuns { + joined = run.waiters == 2 + } + state.mu.Unlock() + if joined { + break + } + time.Sleep(time.Millisecond) + } + require.True(t, joined, "second reader did not join the shared probe") + cancel() + require.ErrorIs(t, awaitResult(t, first), context.Canceled) + fixture.releaseProbe() + got := awaitResult(t, second) + require.NoError(t, got.err) + require.True(t, got.status.Running) + require.Equal(t, int32(1), fixture.requests.Load(), "readers started duplicate readiness requests") +} + +// An explicit port check must wait for an active operation. A nil readiness +// probe keeps this case off the network, so synctest can observe the blocked +// call directly instead of guessing whether it ran during a short sleep. +func TestStatusAtPort_WaitsForActiveOperation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m := testEngineManifest(filepath.Join(t.TempDir(), "missing"+exeExt())) + platform := m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] + platform.Runtime.Ready = nil + m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] = platform + ex := newTestExecutor(t, m) + st, err := ex.state("fake") + require.NoError(t, err) + st.lockOperation() + // Unlock even if an assertion ends the test before the normal release. + var once sync.Once + unlock := func() { once.Do(st.unlockOperation) } + defer unlock() + // Save the result while synctest waits for the caller to block or finish. + finished := make(chan statusResult, 1) + go func() { + status, err := ex.StatusAtPort("fake", 12345) + finished <- statusResult{status, err} + }() + synctest.Wait() + select { + case <-finished: + t.Fatal("explicit port check returned before the operation finished") + default: + } + unlock() + require.NoError(t, awaitResult(t, finished).err) + }) +} + +func TestStartDoesNotWaitForPassiveProbe(t *testing.T) { + fixture := newBlockedReadinessFixture(t) + ex := fixture.executor + st, err := ex.state("fake") + require.NoError(t, err) + st.plat.Detect = []string{fakeEngineBin} + st.plat.Runtime.Bin = fakeEngineBin + finished := make(chan statusResult, 1) + go func() { + status, err := ex.Status("fake") + finished <- statusResult{status, err} + }() + awaitProbeStart(t, fixture.probeStarted) + port, err := freePort() + require.NoError(t, err) + started := make(chan error, 1) + go func() { started <- ex.StartWith(context.Background(), "fake", startOpts{Port: port}) }() + defer func() { fixture.releaseProbe(); _ = ex.Stop("fake") }() + require.NoError(t, awaitResult(t, started)) + fixture.releaseProbe() + got := awaitResult(t, finished) + require.NoError(t, got.err) + require.True(t, got.status.Running) + require.Equal(t, port, got.status.Port) +} + +type statusResult struct { + status EngineStatus + err error +} + +type blockedReadinessFixture struct { + executor *Executor + port int + probeStarted <-chan struct{} + probeCanceled <-chan struct{} + releaseProbe func() + requests *atomic.Int32 +} + +// newBlockedReadinessFixture gives a fake engine a live listener but no local +// executable, forcing Status through passive adoption. The listener holds its +// root readiness reply until released. Health polling is disabled so the tests +// control this one probe. Cleanup releases the reply before closing the server. +func newBlockedReadinessFixture(t *testing.T) blockedReadinessFixture { + t.Helper() + // Notify the test when a readiness request reaches the server. The buffer + // and nonblocking send keep later requests from waiting on this signal. + started := make(chan struct{}, 1) + // Report when cancellation reaches the server, independently of reply release. + canceled := make(chan struct{}, 1) + // Hold replies while the test exercises an overlapping operation or timeout. + // Closing the channel releases every waiting handler, including in cleanup. + release := make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + requests := &atomic.Int32{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + requests.Add(1) + select { + case started <- struct{}{}: + default: + } + select { + case <-release: + w.WriteHeader(http.StatusOK) + case <-r.Context().Done(): + select { + case canceled <- struct{}{}: + default: + } + } + })) + t.Cleanup(func() { + unblock() + srv.Close() + }) + port := portOf(t, srv.URL) + m := testEngineManifest(filepath.Join(t.TempDir(), "missing"+exeExt())) + platform := m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] + platform.Runtime.Port = port + platform.Runtime.Health = nil + m.Platforms[runtime.GOOS+"/"+runtime.GOARCH] = platform + return blockedReadinessFixture{ + executor: newTestExecutor(t, m), + port: port, + probeStarted: started, + probeCanceled: canceled, + releaseProbe: unblock, + requests: requests, + } +} + +// awaitProbeStart bounds the wait for the fixture's readiness request. +func awaitProbeStart(t *testing.T, started <-chan struct{}) { + t.Helper() + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("readiness probe did not start") + } +} + +// awaitResult bounds waits on goroutines so a regression fails the test instead +// of leaving it hung. Channels are buffered or closed so senders cannot block +// after a timeout. +func awaitResult[T any](t *testing.T, ch <-chan T) T { + t.Helper() + select { + case result := <-ch: + return result + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for operation") + var zero T + return zero + } +} From a26759e037d128a14e35abfe58ae1e74a2eb288a Mon Sep 17 00:00:00 2001 From: Jared Ping <13906915+JaredP94@users.noreply.github.com> Date: Mon, 2 Nov 2026 23:20:54 +0200 Subject: [PATCH 2/5] fix(engines): report unmeasured install stages without percentages Adapt NVIDIA/Personal-AI-Router@27b342a: omit percentages for verification, opaque install commands, and the fork's oMLX disk-image copy. Keep measured download progress and terminal outcomes. Desktop clears the previous percentage for both local and remote install stages; the TUI displays stages and failure reasons without fabricated numbers. The broker and remote stream relay optional progress fields verbatim. Regenerate the service contract and document the omission rule. Existing unreleased engine-manager 1.4.0 and TUI 1.2.1 bumps cover these fixes relative to v0.8.0. Co-authored-by: ckelseynv <143756939+ckelseynv@users.noreply.github.com> Signed-off-by: Jared Ping <13906915+JaredP94@users.noreply.github.com> --- .../electron/service-bridge/modular-state.ts | 17 ++- desktop/tests/modular/install-state.test.ts | 48 ++++++- services/nvpair-engine-manager/README.md | 12 +- services/nvpair-engine-manager/executor.go | 24 +++- services/nvpair-engine-manager/install.go | 11 +- services/nvpair-engine-manager/progress.go | 8 +- .../nvpair-engine-manager/progress_test.go | 135 +++++++++++++++++- services/nvpair-tui/README.md | 4 + services/nvpair-tui/ui/engines.go | 8 +- .../nvpair-tui/ui/install_progress_test.go | 26 ++++ services/nvpair-ui-broker/README.md | 3 + 11 files changed, 266 insertions(+), 30 deletions(-) create mode 100644 services/nvpair-tui/ui/install_progress_test.go diff --git a/desktop/src/electron/service-bridge/modular-state.ts b/desktop/src/electron/service-bridge/modular-state.ts index 742802576..10b7b0145 100644 --- a/desktop/src/electron/service-bridge/modular-state.ts +++ b/desktop/src/electron/service-bridge/modular-state.ts @@ -291,6 +291,16 @@ function numberValue(value: JsonValue | undefined): number { return typeof value === 'number' ? value : 0 } +/** + * An install frame's percent, or undefined when the step is indeterminate. The + * engine-manager omits percent (or sends 0) for steps it cannot measure; each + * frame replaces the last, so a missing percent must not carry forward. + */ +function measuredPercent(value: JsonValue | undefined): number | undefined { + const pct = numberValue(value) + return pct > 0 ? pct : undefined +} + function booleanValue(value: JsonValue | undefined): boolean { return typeof value === 'boolean' ? value : false } @@ -2284,11 +2294,10 @@ class ModularBridgeState { ? engineProgressKey({ nodeId, engineType, operation: 'pull', model }) : '' const existingPull = pullKey ? this.activePulls.get(pullKey) : undefined - const framePercent = numberValue(obj.percent) const percent = operation === 'pull' - ? mergePullProgressPercent(framePercent, existingPull?.percent) - : framePercent + ? mergePullProgressPercent(numberValue(obj.percent), existingPull?.percent) + : measuredPercent(obj.percent) const progress: EngineProgress = { engineType, nodeId, @@ -2869,7 +2878,7 @@ class ModularBridgeState { nodeName: node?.name ?? nodeId, operation: 'install', status: stage, - percent: numberValue(obj.percent) + percent: measuredPercent(obj.percent) } emitBridgePush('engines:progress-changed', progress) } diff --git a/desktop/tests/modular/install-state.test.ts b/desktop/tests/modular/install-state.test.ts index 7f501af15..7559ea987 100644 --- a/desktop/tests/modular/install-state.test.ts +++ b/desktop/tests/modular/install-state.test.ts @@ -6,7 +6,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' vi.mock('electron', () => ({ BrowserWindow: { getAllWindows: () => [] } })) vi.mock('@/electron/window', () => ({ createOverviewWindow: vi.fn() })) -import type { EngineProcessStatus } from '@/shared/types/engines' +import type { EngineProcessStatus, EngineProgress } from '@/shared/types/engines' import { subscribePush } from '@/electron/service-bridge/push-bus' import { getModularBridgeState } from '@/electron/service-bridge/modular-state' @@ -46,7 +46,7 @@ describe('LM Studio install state', () => { vi.advanceTimersByTime(120_000) expect(statuses.at(-1)).toBe('installing') - state.applyEngineManagerProgress({ engine: 'lmstudio', stage: 'installing', percent: 75 }) + state.applyEngineManagerProgress({ engine: 'lmstudio', stage: 'installing' }) vi.advanceTimersByTime(30 * 60_000 - 1) expect(statuses.at(-1)).toBe('installing') vi.advanceTimersByTime(1) @@ -57,3 +57,47 @@ describe('LM Studio install state', () => { expect(statuses.at(-1)).toBe('not-installed') }) }) + +describe('install progress percent', () => { + function captureInstallProgress(): EngineProgress[] { + const frames: EngineProgress[] = [] + unsubscribe = subscribePush(event => { + if ( + event.channel === 'engines:progress-changed' && + event.payload.operation === 'install' + ) { + frames.push(event.payload) + } + }) + return frames + } + + it('drops a local install to indeterminate after a measured download', () => { + const state = getModularBridgeState() + const frames = captureInstallProgress() + state.setSelfId('local-node') + + state.applyEngineManagerProgress({ engine: 'lmstudio', stage: 'downloading', percent: 42 }) + expect(frames.at(-1)?.percent).toBe(42) + + state.applyEngineManagerProgress({ engine: 'lmstudio', stage: 'downloading', percent: 100 }) + state.applyEngineManagerProgress({ engine: 'lmstudio', stage: 'installing' }) + expect(frames.at(-1)?.status).toBe('installing') + expect(frames.at(-1)?.percent).toBeUndefined() + }) + + it('drops a remote install to indeterminate after a measured download', () => { + const state = getModularBridgeState() + const frames = captureInstallProgress() + const frame = { node: 'remote-node', engine: 'ollama', op: 'install' } + + state.applyRemoteEngineProgress({ ...frame, stage: 'downloading', percent: 100 }) + expect(frames.at(-1)?.percent).toBe(100) + + state.applyRemoteEngineProgress({ ...frame, stage: 'installing' }) + expect(frames.at(-1)?.status).toBe('installing') + expect(frames.at(-1)?.percent).toBeUndefined() + + state.applyRemoteEngineProgress({ ...frame, stage: 'failed', percent: -1 }) + }) +}) diff --git a/services/nvpair-engine-manager/README.md b/services/nvpair-engine-manager/README.md index d6b0afbe2..271f4e24a 100644 --- a/services/nvpair-engine-manager/README.md +++ b/services/nvpair-engine-manager/README.md @@ -286,18 +286,24 @@ Notifications (service → caller): `engine:ready{version}`, loaded (in-memory) models changes (explicit load/unload, JIT auto-load, or TTL/idle eviction); `models` is the full `engine:models` shape (incl. `loadedByEngine`) so a consumer swaps its whole snapshot, -`engine:install-progress{engine, stage, percent}`, -`engine:pull-progress{engine, op, stage, percent, message}` (live progress for a +`engine:install-progress{engine, stage, percent?, error?}`, +`engine:pull-progress{engine, op, stage, percent?, message}` (live progress for a local model pull driven via `engine:action{action:"pull_model"}` — the local counterpart of `engine:remote-progress`; frames are coalesced to changes in stage/percent, the engine's terminal success surfaces as `stage:"success"`, and a failed pull emits a terminal `stage:"error", percent:-1, message` frame so a UI converges even if its synchronous call already timed out), -`engine:remote-progress{opId, node, engine, op, stage, percent, message}` +`engine:remote-progress{opId, node, engine, op, stage, percent?, message}` (relayed live progress for a remote install/pull), and — for the error pipeline — `errors:report` / `errors:clear` (consumed by `nvpair-errors` via the broker; see below). +Progress `percent` is optional: unmeasured stages omit it (zero internally), +positive values report measured download progress or terminal completion, and +negative values are failure sentinels. Verification, vendor install commands, +and disk-image copies report stages without invented percentages. Clients must +replace the previous install percentage with an indeterminate stage when absent. + For `engine:action` with `action: "pull_model"`, a streaming engine-level failure (including an HTTP 200 NDJSON `{"error":"…"}` frame) returns a JSON-RPC error, not a successful result containing an error body. The peer's diff --git a/services/nvpair-engine-manager/executor.go b/services/nvpair-engine-manager/executor.go index ef4c82037..d9648443e 100644 --- a/services/nvpair-engine-manager/executor.go +++ b/services/nvpair-engine-manager/executor.go @@ -264,16 +264,26 @@ func (st *engineState) pathVars() map[string]string { return vars } -func progress(engine, stage string, pct int) map[string]any { - return map[string]any{"engine": engine, "stage": stage, "percent": pct} -} - // emitInstallProgress reports one install-progress step to both consumers: the // local engine:install-progress notification (this node's UI) and the progress -// hub (so an ec streaming handler can relay it to a remote initiator). +// hub (so an ec streaming handler can relay it to a remote initiator). A pct of +// 0 marks a step with no measurable progress and is omitted from the wire. func (e *Executor) emitInstallProgress(engine, stage string, pct int) { - e.notify("engine:install-progress", progress(engine, stage, pct)) - e.progress.publish(ProgressEvent{Engine: engine, Op: "install", Stage: stage, Percent: pct}) + e.emitInstallFrame(engine, stage, pct, "") +} + +// emitInstallFrame is emitInstallProgress with an optional failure reason, +// carried as "error" on the notification and Message on the hub event. +func (e *Executor) emitInstallFrame(engine, stage string, pct int, errMsg string) { + params := map[string]any{"engine": engine, "stage": stage} + if wirePercentIncluded(pct) { + params["percent"] = pct + } + if errMsg != "" { + params["error"] = errMsg + } + e.notify("engine:install-progress", params) + e.progress.publish(ProgressEvent{Engine: engine, Op: "install", Stage: stage, Percent: pct, Message: errMsg}) } // emitPullProgress reports one model-pull step to both consumers: the local diff --git a/services/nvpair-engine-manager/install.go b/services/nvpair-engine-manager/install.go index bcba61def..7f1e910a7 100644 --- a/services/nvpair-engine-manager/install.go +++ b/services/nvpair-engine-manager/install.go @@ -82,7 +82,7 @@ func (e *Executor) Install(ctx context.Context, engine string) error { // Escape hatch: vendor-script install with no checksum. Logged // loudly so the weaker guarantee is never silent. slog.Warn("running UNPINNED script install (no checksum verification)", "engine", engine) - e.emitInstallProgress(engine, "installing", 50) + e.emitInstallProgress(engine, "installing", 0) argv, err := resolveArgs(inst.Script, vars) if err != nil { e.reportInstallFailed(engine, err) @@ -112,10 +112,10 @@ func (e *Executor) Install(ctx context.Context, engine string) error { } defer os.Remove(dp) vars["download"] = dp - e.emitInstallProgress(engine, "verified", 50) + e.emitInstallProgress(engine, "verified", 0) } if inst.DiskImage != nil { - e.emitInstallProgress(engine, "installing", 75) + e.emitInstallProgress(engine, "installing", 0) id, err := installAppFromDiskImage(ctx, vars["download"], inst.DiskImage) if err != nil { werr := fmt.Errorf("install from disk image failed: %w", err) @@ -127,7 +127,7 @@ func (e *Executor) Install(ctx context.Context, engine string) error { createdID = id } if len(inst.Run) > 0 { - e.emitInstallProgress(engine, "installing", 75) + e.emitInstallProgress(engine, "installing", 0) args, err := resolveArgs(inst.Run, vars) if err != nil { e.reportInstallFailed(engine, err) @@ -469,8 +469,7 @@ func (e *Executor) runCommandTail(ctx context.Context, argv []string, maxLines, } func (e *Executor) reportInstallFailed(engine string, err error) { - e.notify("engine:install-progress", map[string]any{"engine": engine, "stage": "failed", "percent": -1, "error": err.Error()}) - e.progress.publish(ProgressEvent{Engine: engine, Op: "install", Stage: "failed", Percent: -1, Message: err.Error()}) + e.emitInstallFrame(engine, "failed", -1, err.Error()) e.reporter.report(serviceError{ ID: installFailedID(engine), Message: err.Error(), Severity: "error", Action: "retry", EngineType: engine, Operation: "install", diff --git a/services/nvpair-engine-manager/progress.go b/services/nvpair-engine-manager/progress.go index e8c4dd53b..bbe05d56d 100644 --- a/services/nvpair-engine-manager/progress.go +++ b/services/nvpair-engine-manager/progress.go @@ -17,10 +17,10 @@ type ProgressEvent struct { Message string `json:"message,omitempty"` } -// wirePercentIncluded reports whether a pull progress percent should appear on -// the wire. Zero means indeterminate (CLI pulls without byte progress); negative -// values are terminal error sentinels. Install progress still always includes -// percent (including 0) via emitInstallProgress. +// wirePercentIncluded reports whether an install or pull progress percent should +// appear on the wire. Zero means indeterminate (an opaque install command, or a +// CLI pull or download without byte progress); negative values are terminal +// error sentinels. func wirePercentIncluded(pct int) bool { return pct > 0 || pct < 0 } diff --git a/services/nvpair-engine-manager/progress_test.go b/services/nvpair-engine-manager/progress_test.go index a29eaecf9..9674e410f 100644 --- a/services/nvpair-engine-manager/progress_test.go +++ b/services/nvpair-engine-manager/progress_test.go @@ -4,6 +4,13 @@ package main import ( + "context" + "crypto/sha256" + "encoding/hex" + "net/http" + "net/http/httptest" + "path/filepath" + "sync" "testing" "time" ) @@ -65,15 +72,24 @@ func TestProgressHubDropsWhenFull(t *testing.T) { // TestEmitInstallProgressPublishesToHub verifies the executor helper both // notifies and feeds the hub. func TestEmitInstallProgressPublishesToHub(t *testing.T) { - e := &Executor{progress: newProgressHub()} + var gotParams map[string]any + e := &Executor{ + progress: newProgressHub(), + emit: func(_ string, params any) { + gotParams, _ = params.(map[string]any) + }, + } ch, cancel := e.progress.subscribe("ollama") defer cancel() - e.emitInstallProgress("ollama", "installing", 75) + e.emitInstallProgress("ollama", "downloading", 42) + if gotParams["stage"] != "downloading" || gotParams["percent"] != 42 { + t.Fatalf("unexpected notification params: %+v", gotParams) + } select { case ev := <-ch: - if ev.Op != "install" || ev.Stage != "installing" || ev.Percent != 75 { + if ev.Op != "install" || ev.Stage != "downloading" || ev.Percent != 42 { t.Fatalf("unexpected event: %+v", ev) } case <-time.After(time.Second): @@ -81,6 +97,119 @@ func TestEmitInstallProgressPublishesToHub(t *testing.T) { } } +// TestEmitInstallProgressOmitsIndeterminatePercent verifies an unmeasurable +// install step (percent 0) reaches the wire without a percent, so the UI shows +// the stage alone instead of a made-up number. +func TestEmitInstallProgressOmitsIndeterminatePercent(t *testing.T) { + var gotParams map[string]any + e := &Executor{ + progress: newProgressHub(), + emit: func(_ string, params any) { + gotParams, _ = params.(map[string]any) + }, + } + ch, cancel := e.progress.subscribe("ollama") + defer cancel() + + e.emitInstallProgress("ollama", "installing", 0) + + if gotParams["engine"] != "ollama" || gotParams["stage"] != "installing" { + t.Fatalf("unexpected notification params: %+v", gotParams) + } + if _, ok := gotParams["percent"]; ok { + t.Fatalf("indeterminate install progress carried a percent: %+v", gotParams) + } + select { + case ev := <-ch: + if ev.Stage != "installing" || ev.Percent != 0 { + t.Fatalf("unexpected hub event: %+v", ev) + } + case <-time.After(time.Second): + t.Fatal("emitInstallProgress did not publish to the hub") + } +} + +// TestInstallFlowReportsOnlyMeasurablePercents drives a full fetch+run install +// and checks that only measurable (download) and terminal frames carry a +// percent; the verified and installing steps are indeterminate. +func TestInstallFlowReportsOnlyMeasurablePercents(t *testing.T) { + payload := []byte("engine payload") + sum := sha256.Sum256(payload) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write(payload) + })) + defer srv.Close() + bin := filepath.Join(t.TempDir(), "engine"+exeExt()) + m := &Manifest{ + Engine: "fake", DisplayName: "Fake", ManifestVersion: 1, + Platforms: map[string]Platform{hostKey(): { + Detect: []string{bin}, + Install: &Install{ + Fetch: &Fetch{URL: srv.URL, SHA256: hex.EncodeToString(sum[:])}, + Run: []string{fakeEngineBin, "touch", bin}, + Mode: "user", + }, + Runtime: Runtime{Bin: fakeEngineBin}, + }}, + } + + var mu sync.Mutex + var frames []map[string]any + reg := NewRegistry() + reg.engines[m.Engine] = m + ex := NewExecutor(reg, NewReporter(nil), func(method string, params any) { + if method != "engine:install-progress" { + return + } + p, _ := params.(map[string]any) + mu.Lock() + frames = append(frames, p) + mu.Unlock() + }, t.TempDir()) + ex.detectTimeout = 2 * time.Second + + if err := ex.Install(context.Background(), m.Engine); err != nil { + t.Fatalf("install: %v", err) + } + + mu.Lock() + defer mu.Unlock() + stages := map[string]bool{} + measuredDownload := false + for _, f := range frames { + stage, _ := f["stage"].(string) + stages[stage] = true + pct, hasPct := f["percent"] + switch stage { + case "downloading": + if hasPct { + if p, _ := pct.(int); p <= 0 { + t.Errorf("downloading frame carried a non-positive percent: %+v", f) + } + measuredDownload = true + } + case "verified", "installing": + if hasPct { + t.Errorf("%s frame carried a percent: %+v", stage, f) + } + case "done": + if pct != 100 { + t.Errorf("done frame percent = %v, want 100", pct) + } + default: + t.Errorf("unexpected install stage %q: %+v", stage, f) + } + } + if !measuredDownload { + t.Errorf("no downloading frame carried a measured percent; got %+v", frames) + } + for _, want := range []string{"downloading", "verified", "installing", "done"} { + if !stages[want] { + t.Errorf("missing %q frame; got %+v", want, frames) + } + } +} + // TestEmitPullProgressNotifiesAndPublishes verifies the pull helper emits the // local engine:pull-progress notification (with op/message) and also feeds the // hub, so a local pull surfaces progress like a remote pull. diff --git a/services/nvpair-tui/README.md b/services/nvpair-tui/README.md index 9d40bd0f9..9b2c2095c 100644 --- a/services/nvpair-tui/README.md +++ b/services/nvpair-tui/README.md @@ -62,6 +62,10 @@ See [broker supervision](../nvpair-ui-broker/README.md#supervision--recovery). ## Engine operations and workload attribution +Install progress shows the current stage alone when no percentage is measured +(for example verification or a disk-image copy), real download percentages when +available, and the failure reason rather than a negative percentage. + Engine-manager owns support policy. The TUI decodes all seven [`EngineStatus.capabilities`](../nvpair-engine-manager/README.md#effective-capabilities) flags from the target host; missing flags are false. Install, uninstall, pull, diff --git a/services/nvpair-tui/ui/engines.go b/services/nvpair-tui/ui/engines.go index 4936d54a4..3cf3d13c9 100644 --- a/services/nvpair-tui/ui/engines.go +++ b/services/nvpair-tui/ui/engines.go @@ -280,9 +280,15 @@ func (v *enginesView) Update(msg tea.Msg) tea.Cmd { Engine string `json:"engine"` Stage string `json:"stage"` Percent int `json:"percent"` + Error string `json:"error"` } _ = decodeParams(msg.Msg.Params, &p) - v.status = fmt.Sprintf("install %s: %s (%d%%)", p.Engine, p.Stage, p.Percent) + v.status = fmt.Sprintf("install %s: %s", p.Engine, p.Stage) + if p.Stage == "failed" && p.Error != "" { + v.status = fmt.Sprintf("install %s failed: %s", p.Engine, p.Error) + } else if p.Percent > 0 { + v.status += fmt.Sprintf(" (%d%%)", p.Percent) + } case "engine:pull-progress": var p struct { Engine string `json:"engine"` diff --git a/services/nvpair-tui/ui/install_progress_test.go b/services/nvpair-tui/ui/install_progress_test.go new file mode 100644 index 000000000..c7efa3bbf --- /dev/null +++ b/services/nvpair-tui/ui/install_progress_test.go @@ -0,0 +1,26 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package ui + +import ( + "encoding/json" + "testing" + + "nvpair-tui/rpc" +) + +func TestInstallProgressRendersOnlyMeasuredPercent(t *testing.T) { + for _, tc := range []struct{ payload, want string }{ + {`{"engine":"omlx","stage":"installing"}`, "install omlx: installing"}, + {`{"engine":"omlx","stage":"downloading","percent":0}`, "install omlx: downloading"}, + {`{"engine":"omlx","stage":"downloading","percent":42}`, "install omlx: downloading (42%)"}, + {`{"engine":"omlx","stage":"failed","percent":-1,"error":"copy failed"}`, "install omlx failed: copy failed"}, + } { + v := newEnginesView(nil) + v.Update(NotificationMsg{Msg: &rpc.Message{Method: "engine:install-progress", Params: json.RawMessage(tc.payload)}}) + if v.status != tc.want { + t.Errorf("%s: got %q, want %q", tc.payload, v.status, tc.want) + } + } +} diff --git a/services/nvpair-ui-broker/README.md b/services/nvpair-ui-broker/README.md index b18aef6e7..b01b56ceb 100644 --- a/services/nvpair-ui-broker/README.md +++ b/services/nvpair-ui-broker/README.md @@ -406,6 +406,9 @@ ServiceError = { id, message, timestamp, nodeId?, severity?, action?, engineType **Opt-in** (`engine:subscribe`). The notifications `nvpair-engine-manager` emits — `engine:ready`, `engine:state-changed` (full `EngineStatus`), `engine:models-changed` (`{engine, models}`, pushed when an engine's set of models loaded in memory changes), `engine:install-progress`, `engine:pull-progress` (live progress for a local model pull), `engine:remote-progress` (live progress for a remote install/pull) — re-emitted verbatim (their method names are already `engine:`-prefixed). Off by default. +Install progress is relayed without filling absent fields: an omitted `percent` +means an unmeasured stage, not zero percent or the previous download percentage. + #### `connection/cluster-identity` / `connection/cluster-auto-sync` The change-only cluster signals `nvpair-node-settings` pushes when `cluster_id` / `cluster_auto_sync` change. Relayed verbatim and unconditionally — the same way the UI receives them from the settings subprocess directly. From d93c33b3757fd44fb5d395604848a18ffd1bfaef Mon Sep 17 00:00:00 2001 From: Jared Ping <13906915+JaredP94@users.noreply.github.com> Date: Mon, 2 Nov 2026 23:51:32 +0200 Subject: [PATCH 3/5] fix(engine-manager): require live opt-in for non-loopback binds Adapt NVIDIA/Personal-AI-Router#81 with a narrow read-only permission relay for this fork, which has no upstream engine-settings subsystem. Local non-loopback bind overrides require engine_allow_lan_bind, default false, read from node-settings for every request and denied on errors or timeout. Remote starts remain loopback-only. Add persistent typed get/set methods and regressions for denial, revocation, malformed replies, persistence and relay failures. Make existing broker probe cancellation ownership explicit to Go vet, with no behavior change. Bump node-settings to 3.2.0 for the additive setting; existing unreleased engine-manager and broker minor bumps cover their changes relative to v0.8.0. Co-authored-by: Mallikh Kalkere Signed-off-by: Jared Ping <13906915+JaredP94@users.noreply.github.com> --- SECURITY.md | 6 + desktop/docs/service-contract-exceptions.json | 2 + desktop/docs/services-api.md | 7 ++ services/nvpair-engine-manager/README.md | 18 ++- .../bind_validation_test.go | 112 ++++++++++++++++++ services/nvpair-engine-manager/manager.go | 39 ++++-- .../nvpair-engine-manager/settingsrelay.go | 77 ++++++++++++ services/nvpair-node-settings/README.md | 11 ++ services/nvpair-node-settings/manager.go | 30 +++++ services/nvpair-node-settings/manager_test.go | 44 +++++++ services/nvpair-ui-broker/README.md | 7 ++ services/nvpair-ui-broker/broker.go | 4 + .../nvpair-ui-broker/engine_settings_read.go | 54 +++++++++ .../engine_settings_read_test.go | 74 ++++++++++++ services/versions.json | 2 +- 15 files changed, 472 insertions(+), 15 deletions(-) create mode 100644 services/nvpair-engine-manager/bind_validation_test.go create mode 100644 services/nvpair-engine-manager/settingsrelay.go create mode 100644 services/nvpair-ui-broker/engine_settings_read.go create mode 100644 services/nvpair-ui-broker/engine_settings_read_test.go diff --git a/SECURITY.md b/SECURITY.md index 564f683b9..cd92a694b 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -130,6 +130,12 @@ with identity and trust material. It does not automatically protect loopback HTTP, discovery, broker-managed node-info HTTP, OTLP export, Electron IPC, third-party engine APIs, or traffic outside PAIR. +Local engine `bind` overrides require loopback by default. The explicit node +setting `engine_allow_lan_bind` permits non-loopback overrides and is read live, +failing closed if unavailable. It does not authenticate the engine API; remote +starts stay loopback-only. Disabling it does not rebind existing listeners, and +user-authored manifest commands remain the operator's responsibility. + Engine installation executes downloaded vendor software; checksum verification depends on the engine manifest and is not universally required. Catalogs, updates, model downloads, Docker image pulls, engines, and applications can diff --git a/desktop/docs/service-contract-exceptions.json b/desktop/docs/service-contract-exceptions.json index e8ea5cb4c..9f11b8fa0 100644 --- a/desktop/docs/service-contract-exceptions.json +++ b/desktop/docs/service-contract-exceptions.json @@ -1,6 +1,8 @@ { "$comment": "Intentionally unintegrated JSON-RPC methods for desktop ↔ services contract checks. Each ignoredMethods entry needs a current technical reason. Edit by hand; run npm run service-contracts:write to refresh docs/services-api.md after services/ changes.", "ignoredMethods": { + "engine:settings-request": "Worker-to-broker internal live permission read. Only settings/get-engine-allow-lan-bind is allowed; the broker returns a correlated engine:settings-reply directly to the originating worker, never the renderer.", + "engine:settings-reply": "Broker-to-worker internal response for the live LAN-bind permission read. Missing, malformed, failed or timed-out values deny non-loopback local start overrides.", "schedule:priority": "Broker-internal. nvpair-job-scheduler emits a per-engine priority snapshot ({engine, nodes, ranks?} — ordered node ids plus each node's pending count) to nvpair-ui-broker, which forwards it to ollama-proxy and lmstudio-proxy via node/set-priority; each proxy then layers its own optimistic burst reservations on that baseline. PAIR consumes the routing result rather than this notification, and must not mirror the pending/reservation accounting.", "engine:restore-enabled": "Broker-internal startup restoration. nvpair-ui-broker emits engine:restore-enabled directly to its supervised engine-manager after the managed Ollama port gate and on manager respawn; it is not a renderer/UI notification.", "proxy:ready": "Consumed, not missing: the broker relays it and normalizeBrokerProxy (modular-supervisor.ts) strips the `proxy:` prefix, so the bridge handles the de-prefixed `ready` (sets proxyPort). The literal `proxy:ready` is intentionally absent from our TS — extractor limitation, not a gap.", diff --git a/desktop/docs/services-api.md b/desktop/docs/services-api.md index 6190482b1..c41cbfc3c 100644 --- a/desktop/docs/services-api.md +++ b/desktop/docs/services-api.md @@ -31,9 +31,11 @@ - ⚠️ nvpair-job-scheduler → scheduler:tick - ⚠️ nvpair-node-info → n/a - ⚠️ nvpair-node-settings → settings/get-cluster-auto-sync +- ⚠️ nvpair-node-settings → settings/get-engine-allow-lan-bind - ⚠️ nvpair-node-settings → settings/get-force-ports - ⚠️ nvpair-node-settings → settings/get-telemetry-config - ⚠️ nvpair-node-settings → settings/set-cluster-auto-sync +- ⚠️ nvpair-node-settings → settings/set-engine-allow-lan-bind - ⚠️ nvpair-node-settings → settings/set-force-ports - ⚠️ nvpair-ui-broker → discovery:unsubscribe - ⚠️ nvpair-ui-broker → engine:set-reserved-port @@ -42,6 +44,7 @@ - ⚠️ nvpair-ui-broker → metrics:unsubscribe - ⚠️ nvpair-ui-broker → proxy:get-status - ⚠️ nvpair-ui-broker → proxy:unsubscribe +- ⚠️ nvpair-ui-broker → settings/get-engine-allow-lan-bind - ⚠️ nvpair-ui-broker → workloads:unsubscribe - ⚠️ ollama-proxy → node/selected - ⚠️ ollama-proxy → node/set-local-backend @@ -214,11 +217,13 @@ | `settings/get-cluster-auto-sync` | request (we call) | ⚠️ not called | | `settings/get-cluster-friendly-name` | request (we call) | ✅ yes | | `settings/get-cluster-id` | request (we call) | ✅ yes | +| `settings/get-engine-allow-lan-bind` | request (we call) | ⚠️ not called | | `settings/get-force-ports` | request (we call) | ⚠️ not called | | `settings/get-telemetry-config` | request (we call) | ⚠️ not called | | `settings/set-cluster-auto-sync` | request (we call) | ⚠️ not called | | `settings/set-cluster-friendly-name` | request (we call) | ✅ yes | | `settings/set-cluster-id` | request (we call) | ✅ yes | +| `settings/set-engine-allow-lan-bind` | request (we call) | ⚠️ not called | | `settings/set-force-ports` | request (we call) | ⚠️ not called | | `settings/set-telemetry-config` | request (we call) | ✅ yes | @@ -296,6 +301,7 @@ Client surface: requests are sent to the broker; notifications are consumed from | `broker:epoch` | notification (we consume) | ✅ yes | | `discovery:nodes-changed` | notification (we consume) | ✅ yes | | `engine:restore-enabled` | notification (we consume) | ➖ ignored | +| `engine:settings-reply` | notification (we consume) | ➖ ignored | | `errors:clear` | notification (we consume) | ✅ yes | | `errors:report` | notification (we consume) | ✅ yes | | `errors:update` | notification (we consume) | ✅ yes | @@ -334,6 +340,7 @@ Client surface: requests are sent to the broker; notifications are consumed from | `proxy:subscribe` | request (we call) | ✅ yes | | `proxy:unsubscribe` | request (we call) | ⚠️ not called | | `ready` | request (we call) | ✅ yes | +| `settings/get-engine-allow-lan-bind` | request (we call) | ⚠️ not called | | `telemetry/get-status` | request (we call) | ✅ yes | | `telemetry/purge-storage` | request (we call) | ✅ yes | | `telemetry/start` | request (we call) | ✅ yes | diff --git a/services/nvpair-engine-manager/README.md b/services/nvpair-engine-manager/README.md index 271f4e24a..81232dff9 100644 --- a/services/nvpair-engine-manager/README.md +++ b/services/nvpair-engine-manager/README.md @@ -213,7 +213,7 @@ Requests (caller → service): | `engine:remote-install` | `{ node, engine, start? }` | `{ opId, status: EngineStatus }` after the remote install (live progress via `engine:remote-progress`) | | `engine:remote-install-tool` | `{ node, tool }` | `{ opId, result: { tool, installed: true } }`, or the remote tool's typed error | | `engine:remote-pull-model` | `{ node, engine, model?, params? }` | `{ opId, result }` after a successful remote pull (live progress via `engine:remote-progress`); engine-level failures, including HTTP 200 streams containing `error`, return a JSON-RPC error | -| `engine:remote-start` | `{ node, engine, port? }` | `EngineStatus` from the remote node (always the manifest's `runtime.bind`; no per-call bind override on the remote path) | +| `engine:remote-start` | `{ node, engine, port? }` | `EngineStatus` from the remote node (always loopback `127.0.0.1`; no per-call bind override on the remote path) | | `engine:remote-stop` | `{ node, engine }` | `EngineStatus` from the remote node | | `shutdown` | — | `null` | | `log/set-level` | `{ level }` | `{ level }` | @@ -242,6 +242,22 @@ a successful command must leave the tool detectable. | `-32011` | `unsupported-platform`: the node is not `darwin/arm64`. | | `-32012` | `install-failed`: the install command failed or macmon was not detected afterwards. Command failures include only the last 20 stderr lines, capped to the last 4096 bytes; stdout is discarded. | +### Local bind permission + +A local `bind` override must be an IP address on loopback unless the operator +explicitly enables the persisted node setting `engine_allow_lan_bind` (default +false). Engine-manager reads it live through the broker on each non-loopback +override; absent settings workers, read errors, invalid replies and a two-second +timeout all deny permission. Revoking it affects subsequent starts, not listeners +already running. Remote starts always use `127.0.0.1`, regardless of this setting. +This opt-in does not add authentication to third-party engine APIs. It governs +per-call overrides, not user-authored manifest launch commands. + +Internal notifications `engine:settings-request{id, method}` (only +`settings/get-engine-allow-lan-bind`) and +`engine:settings-reply{id, result, error?}` carry the correlated read between +worker and broker; neither is a client subscription event. + ### Passive status and model deadlines Status and get-installed share bounded presence probes without holding the diff --git a/services/nvpair-engine-manager/bind_validation_test.go b/services/nvpair-engine-manager/bind_validation_test.go new file mode 100644 index 000000000..7a1ea47d1 --- /dev/null +++ b/services/nvpair-engine-manager/bind_validation_test.go @@ -0,0 +1,112 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + "testing/synctest" +) + +func TestLiveLANBindSetting(t *testing.T) { + var out bytes.Buffer + ex := NewExecutor(NewRegistry(), NewReporter(nil), nil, t.TempDir()) + m := NewManager(NewCodec(&out), ex, nil) + var result json.RawMessage + var replyError string + reads := 0 + m.settingsRelay.send = func(method string, params any) error { + if method != "engine:settings-request" { + t.Fatalf("unexpected method %s", method) + } + raw, _ := json.Marshal(params) + var request struct{ ID, Method string } + if err := json.Unmarshal(raw, &request); err != nil { + return err + } + if request.Method != "settings/get-engine-allow-lan-bind" { + t.Fatalf("unexpected read %s", request.Method) + } + reads++ + reply, _ := json.Marshal(settingsReply{ID: request.ID, Result: result, Error: replyError}) + m.handleMessage(context.Background(), &Message{Method: "engine:settings-reply", Params: reply}) + return nil + } + for _, tc := range []struct { + result, err string + allowed bool + }{ + {`{"value":false}`, "", false}, + {`{"value":true}`, "", true}, + {`{"value":false}`, "", false}, + {`{}`, "", false}, + {`{"value":null}`, "", false}, + {`{"value":"true"}`, "", false}, + {`{"value":true}`, "unavailable", false}, + } { + result, replyError = json.RawMessage(tc.result), tc.err + out.Reset() + id := json.RawMessage("1") + m.runOp(context.Background(), &Message{ID: &id, Method: "engine:start", Params: json.RawMessage(`{"engine":"missing","bind":"0.0.0.0"}`)}) + denied := strings.Contains(out.String(), "bind must be a loopback address") + if denied == tc.allowed { + t.Errorf("result=%s err=%s allowed=%v response=%s", tc.result, tc.err, tc.allowed, out.String()) + } + } + if reads != 7 { + t.Fatalf("read count=%d, want a live read each time", reads) + } + m.settingsRelay.send = func(string, any) error { return errors.New("broker unavailable") } + if m.lanBindAllowed(context.Background()) { + t.Fatal("transport error allowed LAN bind") + } +} + +func TestLANBindReadTimeoutFailsClosed(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var out bytes.Buffer + m := NewManager(NewCodec(&out), nil, nil) + m.settingsRelay.send = func(string, any) error { return nil } + if m.lanBindAllowed(context.Background()) { + t.Fatal("timeout allowed LAN bind") + } + if len(m.settingsRelay.pending) != 0 { + t.Fatal("timeout leaked pending read") + } + }) +} + +func TestLoopbackBindDoesNotReadSetting(t *testing.T) { + for _, bind := range []string{"127.0.0.1", "127.0.0.2", "::1", "::ffff:127.0.0.1", ""} { + var out bytes.Buffer + ex := NewExecutor(NewRegistry(), NewReporter(nil), nil, t.TempDir()) + m := NewManager(NewCodec(&out), ex, nil) + m.settingsRelay.send = func(string, any) error { t.Fatal("loopback required setting lookup"); return nil } + id := json.RawMessage("1") + m.runOp(context.Background(), &Message{ID: &id, Method: "engine:start", Params: json.RawMessage(fmt.Sprintf(`{"engine":"missing","bind":%q}`, bind))}) + if !strings.Contains(out.String(), "unknown engine") { + t.Fatalf("unexpected response %s", out.String()) + } + } +} + +func TestRunOpRejectsNonLoopbackBindByDefault(t *testing.T) { + for _, bind := range []string{"0.0.0.0", "192.168.1.5", "::", "::ffff:192.168.1.5"} { + var out bytes.Buffer + ex := NewExecutor(NewRegistry(), NewReporter(nil), nil, t.TempDir()) + m := NewManager(NewCodec(&out), ex, nil) + ctx, cancel := context.WithCancel(context.Background()) + cancel() // An unavailable setting read must fail closed. + id := json.RawMessage("1") + m.runOp(ctx, &Message{ID: &id, Method: "engine:start", Params: json.RawMessage(fmt.Sprintf(`{"engine":"fake","bind":%q}`, bind))}) + if !strings.Contains(out.String(), "bind must be a loopback address") { + t.Errorf("bind %s was not rejected: %s", bind, out.String()) + } + } +} diff --git a/services/nvpair-engine-manager/manager.go b/services/nvpair-engine-manager/manager.go index a92d164a3..117263cab 100644 --- a/services/nvpair-engine-manager/manager.go +++ b/services/nvpair-engine-manager/manager.go @@ -62,11 +62,12 @@ type setPortParam struct { // start, stop, restart, action) run in their own goroutine so the read // loop never blocks; the codec serializes the concurrent responses. type Manager struct { - codec *Codec - exec *Executor - peers *peerDirectory - addrs *reach.Chooser - mesh *clustertrust.Mesh // cluster identity + pins for dialing peers' ec surfaces + codec *Codec + settingsRelay settingsRelay + exec *Executor + peers *peerDirectory + addrs *reach.Chooser + mesh *clustertrust.Mesh // cluster identity + pins for dialing peers' ec surfaces // remoteHTTP / readyHTTP are long-lived per-peer mTLS pools. A throwaway // Transport per engine:remote-* call leaked the idle socket. readyHTTP // uses the longer header budget for start/delete (see waitsForEngineReadiness). @@ -80,11 +81,12 @@ func NewManager(codec *Codec, exec *Executor, mesh *clustertrust.Mesh) *Manager // handleMessage is reached before Run() installs the real CancelFunc // (e.g. a unit test calling it directly); Run overwrites it. return &Manager{ - codec: codec, - exec: exec, - peers: newPeerDirectory(), - addrs: reach.NewChooser(), - mesh: mesh, + codec: codec, + settingsRelay: settingsRelay{send: codec.Notify}, + exec: exec, + peers: newPeerDirectory(), + addrs: reach.NewChooser(), + mesh: mesh, remoteHTTP: clustertrust.NewPeerClientPoolOpts(mesh, clustertrust.PeerClientOptions{ ResponseHeaderTimeout: remoteResponseHeaderTimeout, }), @@ -168,6 +170,10 @@ func (m *Manager) handleMessage(ctx context.Context, msg *Message) { return } + if msg.Method == "engine:settings-reply" && msg.IsNotification() { + m.settingsRelay.reply(msg.Params) + return + } if msg.Method == noderec.NotifyNodes { m.applyDiscovery(msg) return @@ -298,9 +304,16 @@ func (m *Manager) runOp(ctx context.Context, msg *Message) { m.codec.RespondError(msg.ID, -32602, "port must be between 0 and 65535") return } - if p.Bind != "" && net.ParseIP(p.Bind) == nil { - m.codec.RespondError(msg.ID, -32602, "bind must be a valid IP address") - return + if p.Bind != "" { + ip := net.ParseIP(p.Bind) + if ip == nil { + m.codec.RespondError(msg.ID, -32602, "bind must be a valid IP address") + return + } + if !ip.IsLoopback() && !m.lanBindAllowed(ctx) { + m.codec.RespondError(msg.ID, -32602, "bind must be a loopback address unless engine_allow_lan_bind is enabled") + return + } } start := func() error { return m.exec.StartWith(ctx, p.Engine, startOpts{Port: p.Port, Bind: p.Bind}) diff --git a/services/nvpair-engine-manager/settingsrelay.go b/services/nvpair-engine-manager/settingsrelay.go new file mode 100644 index 000000000..c8f0d94e5 --- /dev/null +++ b/services/nvpair-engine-manager/settingsrelay.go @@ -0,0 +1,77 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "encoding/json" + "sync" + "time" +) + +// Only this read is relayed upward; the worker cannot mutate node settings. +const lanBindSettingMethod = "settings/get-engine-allow-lan-bind" + +type settingsReply struct { + ID string `json:"id"` + Result json.RawMessage `json:"result"` + Error string `json:"error,omitempty"` +} + +type settingsRelay struct { + mu sync.Mutex + pending map[string]chan settingsReply + send func(string, any) error +} + +// lanBindAllowed reads through on each non-loopback override. Missing settings, +// malformed values, transport failures and timeouts all deny permission. +func (m *Manager) lanBindAllowed(parent context.Context) bool { + ctx, cancel := context.WithTimeout(parent, 2*time.Second) + defer cancel() + if ctx.Err() != nil { + return false + } + r := &m.settingsRelay + id := newOpID() + ch := make(chan settingsReply, 1) + r.mu.Lock() + if len(r.pending) >= 64 { + r.mu.Unlock() + return false + } + if r.pending == nil { + r.pending = make(map[string]chan settingsReply) + } + r.pending[id] = ch + r.mu.Unlock() + defer func() { r.mu.Lock(); delete(r.pending, id); r.mu.Unlock() }() + if err := r.send("engine:settings-request", map[string]string{"id": id, "method": lanBindSettingMethod}); err != nil { + return false + } + select { + case reply := <-ch: + var result struct { + Value *bool `json:"value"` + } + return reply.Error == "" && json.Unmarshal(reply.Result, &result) == nil && result.Value != nil && *result.Value + case <-ctx.Done(): + return false + } +} + +func (r *settingsRelay) reply(raw json.RawMessage) { + var reply settingsReply + if json.Unmarshal(raw, &reply) != nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if ch := r.pending[reply.ID]; ch != nil { + select { + case ch <- reply: + default: + } + } +} diff --git a/services/nvpair-node-settings/README.md b/services/nvpair-node-settings/README.md index 48adce6c3..79deb5266 100644 --- a/services/nvpair-node-settings/README.md +++ b/services/nvpair-node-settings/README.md @@ -201,6 +201,17 @@ configuration returns the existing snapshot without saving, incrementing its revision, or notifying. Save failures return `-32603` without changing the in-memory configuration or revision. +### Engine LAN-bind opt-in + +`settings/get-engine-allow-lan-bind` returns `{"value": false}` by default. +`settings/set-engine-allow-lan-bind {"value": true|false}` persists +`engine_allow_lan_bind` and returns `{"ok": true}` only after a successful save. +Engine-manager reads this setting live for non-loopback local `bind` overrides; +read failures deny permission. This deliberately exposes an unauthenticated +third-party engine API when enabled; use only on networks you control. Remote +starts remain loopback-only. No desktop toggle is provided; operators can use +these JSON-RPC methods. Disabling does not rebind an already-running listener. + `settings/get-cluster-id` returns the raw stored `cluster_id` string — the same shape the other non-telemetry getters use (`{"value": }`) — so the getter round-trips with diff --git a/services/nvpair-node-settings/manager.go b/services/nvpair-node-settings/manager.go index f710a92f9..b897df7cd 100644 --- a/services/nvpair-node-settings/manager.go +++ b/services/nvpair-node-settings/manager.go @@ -73,6 +73,10 @@ type Settings struct { // enough. ClusterFriendlyName string `json:"cluster_friendly_name"` + // EngineAllowLANBind explicitly permits non-loopback local start overrides. + // Engine-manager reads it live; absent or unreadable settings deny access. + EngineAllowLANBind bool `json:"engine_allow_lan_bind"` + // TelemetryEnabled controls whether proxies emit OpenTelemetry spans. TelemetryEnabled bool `json:"telemetry_enabled"` @@ -592,6 +596,32 @@ func (m *Manager) handleMessage(msg *Message) { slog.Warn("failed to emit telemetry-changed notification after set", "err", err) } + case "settings/get-engine-allow-lan-bind": + m.mu.RLock() + v := m.settings.EngineAllowLANBind + m.mu.RUnlock() + m.codec.Respond(msg.ID, map[string]bool{"value": v}) + + case "settings/set-engine-allow-lan-bind": + var p boolValueParams + if err := json.Unmarshal(msg.Params, &p); err != nil || p.Value == nil { + m.codec.RespondError(msg.ID, -32602, `invalid params: expected {"value": }`) + return + } + m.mu.RLock() + newSettings := m.settings + m.mu.RUnlock() + newSettings.EngineAllowLANBind = *p.Value + if err := m.saveSettings(newSettings); err != nil { + slog.Error("failed to persist engine-allow-lan-bind", "err", err) + m.codec.RespondError(msg.ID, -32603, "failed to persist setting: "+err.Error()) + return + } + m.mu.Lock() + m.settings = newSettings + m.mu.Unlock() + m.codec.Respond(msg.ID, map[string]bool{"ok": true}) + case "shutdown": if err := m.codec.Respond(msg.ID, nil); err != nil { log.Printf("failed to respond to shutdown: %v", err) diff --git a/services/nvpair-node-settings/manager_test.go b/services/nvpair-node-settings/manager_test.go index 368c70284..0bc1985a4 100644 --- a/services/nvpair-node-settings/manager_test.go +++ b/services/nvpair-node-settings/manager_test.go @@ -232,6 +232,50 @@ func TestClusterAutoSyncRoundTrip(t *testing.T) { } } +func TestEngineAllowLANBindRoundTrip(t *testing.T) { + m, rw, path := newTestManager(t) + + got := callAndDecode[map[string]bool](t, m, rw, 1, "settings/get-engine-allow-lan-bind", nil) + if got["value"] != false { + t.Fatalf("default engine-allow-lan-bind = %v, want false", got["value"]) + } + + _ = callAndDecode[map[string]bool](t, m, rw, 2, "settings/set-engine-allow-lan-bind", map[string]bool{"value": true}) + got = callAndDecode[map[string]bool](t, m, rw, 3, "settings/get-engine-allow-lan-bind", nil) + if got["value"] != true { + t.Fatalf("after set, value = %v, want true", got["value"]) + } + reloadedRW := newCaptureRW() + reloaded, err := NewManager(NewCodec(reloadedRW), path) + if err != nil { + t.Fatal(err) + } + persisted := callAndDecode[map[string]bool](t, reloaded, reloadedRW, 1, "settings/get-engine-allow-lan-bind", nil) + if !persisted["value"] { + t.Fatal("LAN bind permission did not persist") + } + for _, raw := range []string{`{}`, `{"value":null}`, `{"value":"true"}`} { + callExpectError(t, m, rw, 4, "settings/set-engine-allow-lan-bind", json.RawMessage(raw), -32602) + } + if err := os.Mkdir(path+".tmp", 0o700); err != nil { + t.Fatal(err) + } + callExpectError(t, m, rw, 4, "settings/set-engine-allow-lan-bind", json.RawMessage(`{"value":false}`), -32603) + got = callAndDecode[map[string]bool](t, m, rw, 5, "settings/get-engine-allow-lan-bind", nil) + if !got["value"] { + t.Fatal("failed save changed live permission") + } + if err := os.Remove(path + ".tmp"); err != nil { + t.Fatal(err) + } + + _ = callAndDecode[map[string]bool](t, m, rw, 4, "settings/set-engine-allow-lan-bind", map[string]bool{"value": false}) + got = callAndDecode[map[string]bool](t, m, rw, 5, "settings/get-engine-allow-lan-bind", nil) + if got["value"] != false { + t.Fatalf("after unset, value = %v, want false", got["value"]) + } +} + // TestClusterIDRoundTrip locks down the basic get/set contract for // the cluster identifier. Empty is the default; the round-trip is // byte-for-byte (no normalization, no trimming) so a future diff --git a/services/nvpair-ui-broker/README.md b/services/nvpair-ui-broker/README.md index b01b56ceb..c94c943f6 100644 --- a/services/nvpair-ui-broker/README.md +++ b/services/nvpair-ui-broker/README.md @@ -703,6 +703,13 @@ Opt into / out of the `engine:` stream (off by default). Acks `{ subscrib Any other `engine:*` request is forwarded to `nvpair-engine-manager` verbatim and its response relayed straight back. This covers the whole engine control plane: `engine:get-installed`, `engine:describe`, `engine:status`, `engine:install`, `engine:uninstall`, `engine:start`, `engine:stop`, `engine:restart`, `engine:set-port`, `engine:action`, `engine:logs`, `engine:errors`, `engine:models`. (`engine:set-port` persists the engine's server port as a manifest override that survives a restart — see the `nvpair-engine-manager` README.) Lifecycle ops run for minutes (reporting progress via the `engine:install-progress` / `engine:state-changed` push events), so the relay imposes **no broker-side timeout** — fire the request and watch the event stream for the outcome. Error `-32000 "engine-manager not available"` when no engine-manager is supervised. +Engine-manager's internal `engine:settings-request{id, method}` is handled +separately from subscriptions. Only `settings/get-engine-allow-lan-bind` is +forwarded to node-settings, with a two-second bound, and the correlated +`engine:settings-reply{id, result, error?}` goes directly to the captured worker. +It is read-only and is never broadcast to desktop or TUI clients. Missing workers +and failed reads deny the non-loopback override. + #### `settings/` (generic relay) Any `settings/*` request is forwarded to `nvpair-node-settings` and its response relayed back: `settings/get-force-ports`, `settings/set-force-ports`, `settings/get-cluster-id`, `settings/set-cluster-id`, `settings/get-cluster-auto-sync`, `settings/set-cluster-auto-sync`, `settings/get-cluster-friendly-name`, `settings/set-cluster-friendly-name`, `settings/get-telemetry-config`, `settings/set-telemetry-config`. Error `-32000 "node-settings not available"` when no settings worker is supervised. diff --git a/services/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go index 2923e62da..cbfc1491b 100644 --- a/services/nvpair-ui-broker/broker.go +++ b/services/nvpair-ui-broker/broker.go @@ -1383,6 +1383,10 @@ func (b *Broker) pushSchedulingDemand(active bool) { // pipeline; the rest of each worker's notification stream is logged and // dropped until its control-plane relay is wired in. func (b *Broker) forwardEngineNotification(method string, params json.RawMessage) { + if method == "engine:settings-request" { + b.handleEngineSettingsRead(params) + return + } if b.dispatchErrorsNotif("engine-manager", method, params) { return } diff --git a/services/nvpair-ui-broker/engine_settings_read.go b/services/nvpair-ui-broker/engine_settings_read.go new file mode 100644 index 000000000..bdcb4974d --- /dev/null +++ b/services/nvpair-ui-broker/engine_settings_read.go @@ -0,0 +1,54 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "encoding/json" + "time" +) + +// handleEngineSettingsRead services the worker's narrow live permission read. +// Capture the worker before dispatch so a response never crosses a respawn. +func (b *Broker) handleEngineSettingsRead(raw json.RawMessage) { + var request struct { + ID string `json:"id"` + Method string `json:"method"` + } + if json.Unmarshal(raw, &request) != nil || request.ID == "" { + return + } + em := b.getEngineMgr() + if em == nil { + return + } + settings := b.getSettings() + go func() { + var reply struct { + ID string `json:"id"` + Result json.RawMessage `json:"result"` + Error string `json:"error,omitempty"` + } + reply.ID = request.ID + switch { + case request.Method != "settings/get-engine-allow-lan-bind": + reply.Error = "unsupported engine setting read" + case settings == nil: + reply.Error = "node-settings unavailable" + default: + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + result, rpcErr, err := settings.Call(ctx, request.Method, nil) + switch { + case err != nil: + reply.Error = "node-settings read failed" + case rpcErr != nil: + reply.Error = "node-settings read rejected" + default: + reply.Result = result + } + } + _ = em.Notify("engine:settings-reply", reply) + }() +} diff --git a/services/nvpair-ui-broker/engine_settings_read_test.go b/services/nvpair-ui-broker/engine_settings_read_test.go new file mode 100644 index 000000000..a6fcd0050 --- /dev/null +++ b/services/nvpair-ui-broker/engine_settings_read_test.go @@ -0,0 +1,74 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "testing" + "time" +) + +func TestEngineSettingReadRelay(t *testing.T) { + for _, mode := range []string{"true", "false", "missing", "rpc-error", "unsupported"} { + t.Run(mode, func(t *testing.T) { + engine, replies := newTestRPCWorkerPipe(t) + b := &Broker{} + b.setEngineMgr(engine) + method := "settings/get-engine-allow-lan-bind" + if mode == "unsupported" { + method = "settings/set-engine-allow-lan-bind" + } + if mode != "missing" && mode != "unsupported" { + settings, requests := newTestRPCWorkerPipe(t) + b.setSettings(settings) + go func() { + request, err := requests.Read() + if err != nil { + return + } + if request.Method != method { + _ = requests.RespondError(request.ID, -32601, "wrong method") + return + } + if mode == "rpc-error" { + _ = requests.RespondError(request.ID, -32603, "read failed") + return + } + _ = requests.Respond(request.ID, map[string]bool{"value": mode == "true"}) + }() + } + raw, _ := json.Marshal(map[string]string{"id": "read-1", "method": method}) + b.forwardEngineNotification("engine:settings-request", raw) + done := make(chan *Message, 1) + go func() { msg, _ := replies.Read(); done <- msg }() + select { + case msg := <-done: + if msg == nil || msg.Method != "engine:settings-reply" { + t.Fatalf("unexpected reply %+v", msg) + } + var reply struct { + ID string + Result json.RawMessage + Error string + } + if err := json.Unmarshal(msg.Params, &reply); err != nil { + t.Fatal(err) + } + if reply.ID != "read-1" { + t.Fatalf("uncorrelated reply %+v", reply) + } + if mode == "true" || mode == "false" { + var value struct{ Value bool } + if json.Unmarshal(reply.Result, &value) != nil || value.Value != (mode == "true") || reply.Error != "" { + t.Fatalf("bad result %+v", reply) + } + } else if reply.Error == "" { + t.Fatal("missing failure") + } + case <-time.After(3 * time.Second): + t.Fatal("setting read relay did not finish") + } + }) + } +} diff --git a/services/versions.json b/services/versions.json index ef9094637..a1711d886 100644 --- a/services/versions.json +++ b/services/versions.json @@ -10,7 +10,7 @@ "nvpair-manual-nodes": "1.2.2", "nvpair-workload-manager": "1.3.1", "nvpair-errors": "1.1.1", - "nvpair-node-settings": "3.1.0", + "nvpair-node-settings": "3.2.0", "nvpair-ui-broker": "2.5.0", "nvpair-engine-manager": "1.4.0", "nvpair-cluster-manager": "2.1.1", From b950ace091b678ef2d2a0b37ccbe2c7915348dbe Mon Sep 17 00:00:00 2001 From: Jared Ping <13906915+JaredP94@users.noreply.github.com> Date: Mon, 2 Nov 2026 23:57:25 +0200 Subject: [PATCH 4/5] fix(engine-manager): join one concurrent engine shutdown sweep Port NVIDIA/Personal-AI-Router@72d5605: prepare-shutdown and EOF must wait for one sweep rather than probing foreign listeners twice. Preserve this fork's concurrent engine stops and retry only failed process or command stops that can succeed on a later call. Cover sweep deduplication using lifecycle revisions rather than elapsed-time ratios, concurrent joining, and retry after a failed stop command. The existing unreleased engine-manager 1.4.0 bump covers this fix relative to v0.8.0. Co-authored-by: Terve Signed-off-by: Jared Ping <13906915+JaredP94@users.noreply.github.com> --- services/nvpair-engine-manager/README.md | 8 + services/nvpair-engine-manager/executor.go | 9 +- services/nvpair-engine-manager/lifecycle.go | 80 ++++++- .../shutdown_once_test.go | 197 ++++++++++++++++++ .../testdata/fakeengine/main.go | 10 + services/nvpair-ui-broker/README.md | 4 + 6 files changed, 296 insertions(+), 12 deletions(-) create mode 100644 services/nvpair-engine-manager/shutdown_once_test.go diff --git a/services/nvpair-engine-manager/README.md b/services/nvpair-engine-manager/README.md index 81232dff9..ab2fe48a0 100644 --- a/services/nvpair-engine-manager/README.md +++ b/services/nvpair-engine-manager/README.md @@ -242,6 +242,14 @@ a successful command must leave the tool detectable. | `-32011` | `unsupported-platform`: the node is not `darwin/arm64`. | | `-32012` | `install-failed`: the install command failed or macmon was not detected afterwards. Command failures include only the last 20 stderr lines, capped to the last 4096 bytes; stdout is discarded. | +### Shutdown sweep + +`engine:prepare-shutdown` and the EOF/transport-exit path join one concurrent +engine sweep per executor. Every caller waits for it to finish. Later calls +retry only process-handle or command-stop failures that can still be resolved; +a declined foreign listener is not repeatedly probed. Shutdown rejects new starts +and never changes saved ON/OFF intent. Engines still stop concurrently. + ### Local bind permission A local `bind` override must be an IP address on loopback unless the operator diff --git a/services/nvpair-engine-manager/executor.go b/services/nvpair-engine-manager/executor.go index d9648443e..b6ad7cf06 100644 --- a/services/nvpair-engine-manager/executor.go +++ b/services/nvpair-engine-manager/executor.go @@ -141,8 +141,13 @@ type Executor struct { reservedPort atomic.Int32 // StopAll is terminal for an Executor; the gate closes its start/snapshot race. shuttingDown atomic.Bool - mu sync.Mutex - engines map[string]*engineState + // All shutdown entry points join one concurrent sweep. Only actionable + // process/command stop failures are retained for a later retry. + stopAllOnce sync.Once + retryMu sync.Mutex + failedStops []string + mu sync.Mutex + engines map[string]*engineState } func NewExecutor(reg *Registry, reporter *Reporter, emit func(string, any), baseDir string) *Executor { diff --git a/services/nvpair-engine-manager/lifecycle.go b/services/nvpair-engine-manager/lifecycle.go index a36369da5..0c2f829d9 100644 --- a/services/nvpair-engine-manager/lifecycle.go +++ b/services/nvpair-engine-manager/lifecycle.go @@ -701,7 +701,36 @@ func (e *Executor) Restart(ctx context.Context, engine string) error { // StopAll rejects new starts and terminates every running engine during // shutdown without changing the user's saved ON/OFF intent. +// +// Two entry points ask for it on an ordinary quit — the +// engine:prepare-shutdown handler, which the broker calls from its own +// teardown, and the stdin-EOF path in Run, which follows when the broker then +// closes stdin — because each is the right trigger for a different way of +// being shut down: a broker ordering its teardown, and a parent that simply +// severed the pipe. +// +// One sweep runs per process, and every caller returns only once it has +// finished. A caller that returned while the sweep was still running would +// report engines stopped before they were: the broker would close stdin, +// engine-manager would exit, and the engines it launched would be orphaned. +// Repeating the sweep would re-pay doStop's readiness probe for every adopted +// engine whose stop it declines, each time. +// +// A later caller does retry the engines whose stop failed through a process +// handle or a stop command, such as an `lms server stop` that errored, because +// another attempt can succeed where a declined stop cannot. func (e *Executor) StopAll() { + swept := false + e.stopAllOnce.Do(func() { + swept = true + e.stopAllNow() + }) + if !swept { + e.retryFailedStops() + } +} + +func (e *Executor) stopAllNow() { e.shuttingDown.Store(true) e.mu.Lock() names := make([]string, 0, len(e.engines)) @@ -709,24 +738,47 @@ func (e *Executor) StopAll() { names = append(names, n) } e.mu.Unlock() - // Stop engines concurrently. Each doStop can wait up to the manifest grace - // (default 5s) for its engine to exit, so a sequential sweep would take - // sum(grace) across engines and let a slow first engine delay — or, if the - // parent severs us mid-sweep, entirely prevent — every later engine from - // even being signaled. Running them in parallel bounds the whole sweep to - // ~one grace and guarantees every engine gets its stop signal immediately. - // Each engine still serializes on its own opMu, so this is safe. + e.failedStops = e.stopEngines(names) +} + +// retryFailedStops tries again to stop the engines whose stop failed in the +// sweep. Callers take turns, so two retries never stop one engine at once. +func (e *Executor) retryFailedStops() { + e.retryMu.Lock() + defer e.retryMu.Unlock() + if len(e.failedStops) == 0 { + return + } + e.failedStops = e.stopEngines(e.failedStops) +} + +// stopEngines stops the named engines and returns those whose stop failed +// through a process handle or a stop command, the failures a later attempt can +// fix. Without either, doStop declines a foreign listener or has already +// escalated against PAIR's own orphan, and a retry would only repeat that. +// +// Engines stop concurrently. Each doStop can wait up to the manifest grace +// (default 5s) for its engine to exit, so a sequential sweep would take +// sum(grace) across engines and let a slow first engine delay — or, if the +// parent severs us mid-sweep, entirely prevent — every later engine from even +// being signaled. Running them in parallel bounds the whole sweep to ~one grace +// and guarantees every engine gets its stop signal immediately. Each engine +// still serializes on its own opMu, so this is safe. +func (e *Executor) stopEngines(names []string) []string { var wg sync.WaitGroup - for _, n := range names { + // Each goroutine writes only its own index, so no lock is needed. + retry := make([]bool, len(names)) + for i, n := range names { st, err := e.state(n) if err != nil { continue } wg.Add(1) - go func(st *engineState, n string) { + go func(i int, st *engineState, n string) { defer wg.Done() st.mu.Lock() cancel := st.startCancel + retryable := st.plat.Runtime.modeOrDefault() == "command" || st.proc != nil st.mu.Unlock() if cancel != nil { cancel() @@ -735,10 +787,18 @@ func (e *Executor) StopAll() { defer st.unlockOperation() if err := e.doStop(st, n); err != nil { slog.Warn("engine stop during shutdown failed", "engine", n, "err", err) + retry[i] = retryable } - }(st, n) + }(i, st, n) } wg.Wait() + var failed []string + for i, n := range names { + if retry[i] { + failed = append(failed, n) + } + } + return failed } func (e *Executor) runHealth(ctx context.Context, st *engineState, engine string, port int, gen int64) { diff --git a/services/nvpair-engine-manager/shutdown_once_test.go b/services/nvpair-engine-manager/shutdown_once_test.go new file mode 100644 index 000000000..7b10d2ab8 --- /dev/null +++ b/services/nvpair-engine-manager/shutdown_once_test.go @@ -0,0 +1,197 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "net/url" + "path/filepath" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// adoptedEngineOnLivePort seeds an engine that this process did not launch and +// cannot stop: running, adopted, no owned proc, and a real listener on its port +// so doStop's readiness probe can never confirm it went away. That is the one +// shape whose stop is refused, leaving st.running true — so a repeated sweep +// does the whole thing again instead of hitting the !st.running early return. +// Its bin path is outside the managed install dir, so the port-reclaim branch +// declines it rather than terminating a stranger's process. +func adoptedEngineOnLivePort(t *testing.T, ex *Executor) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err, "listen") + t.Cleanup(func() { _ = ln.Close() }) + + st, err := ex.state("fake") + require.NoError(t, err) + st.mu.Lock() + st.running = true + st.adopted = true + st.proc = nil + st.port = ln.Addr().(*net.TCPAddr).Port + st.binPath = filepath.Join(t.TempDir(), "someone-elses-engine") + st.mu.Unlock() +} + +// TestStopAll_RunsOneSweepPerProcess is the quit-latency guard. An ordinary quit +// asks for StopAll twice — the broker's engine:prepare-shutdown during its +// teardown, and the stdin-EOF path in Run once the broker closes stdin. An +// engine whose stop can only be declined stays running, so each repeat sweep +// would re-pay doStop's readiness probe for it. +func TestStopAll_RunsOneSweepPerProcess(t *testing.T) { + ex := newTestExecutor(t, testEngineManifest(fakeEngineBin)) + adoptedEngineOnLivePort(t, ex) + + ex.StopAll() + st, err := ex.state("fake") + require.NoError(t, err) + st.mu.Lock() + revision := st.revision + require.True(t, st.running, "foreign engine must remain running") + st.mu.Unlock() + + // Count lifecycle transitions, not elapsed time: repeating the sweep must + // not re-enter the engine operation or probe its foreign listener again. + ex.StopAll() + ex.StopAll() + st.mu.Lock() + got := st.revision + st.mu.Unlock() + require.Equal(t, revision, got, "further StopAll calls repeated the sweep") +} + +// TestStopAll_WaitsForTheSweepInFlight is the other half, and the reason this +// is a joining guard rather than a plain skip. A caller that returned while the +// first sweep was still running would report engines stopped before they were: +// the broker would close stdin, engine-manager would exit, and the engines it +// launched would be left running with nothing owning them. +func TestStopAll_WaitsForTheSweepInFlight(t *testing.T) { + ex := newTestExecutor(t, testEngineManifest(fakeEngineBin)) + adoptedEngineOnLivePort(t, ex) + st, err := ex.state("fake") + require.NoError(t, err) + + // Holding the engine's opMu parks the sweep before doStop, so it cannot + // finish until the test lets it, however the goroutines are scheduled. + st.opMu.Lock() + releaseSweep := sync.OnceFunc(st.opMu.Unlock) + t.Cleanup(releaseSweep) + sweepDone := make(chan struct{}) + go func() { + defer close(sweepDone) + ex.StopAll() + }() + + // Join behind the first caller only once it owns the sweep: only the sweep + // sets shuttingDown, and it does so before stopping anything. + require.Eventually(t, ex.shuttingDown.Load, 5*time.Second, 5*time.Millisecond, + "the first StopAll caller never started the sweep") + secondDone := make(chan struct{}) + go func() { + defer close(secondDone) + ex.StopAll() + }() + + require.Never(t, func() bool { return isClosed(secondDone) }, 200*time.Millisecond, 10*time.Millisecond, + "the second caller returned while the first sweep was still running") + + releaseSweep() + require.Eventually(t, func() bool { return isClosed(sweepDone) }, 10*time.Second, 10*time.Millisecond, + "the first StopAll caller never returned once the sweep was released") + require.Eventually(t, func() bool { return isClosed(secondDone) }, 10*time.Second, 10*time.Millisecond, + "the second StopAll caller never returned once the sweep was released") +} + +// isClosed reports whether done has been closed, without blocking. +func isClosed(done <-chan struct{}) bool { + select { + case <-done: + return true + default: + return false + } +} + +// commandDaemon stands in for a command-mode engine's daemon: it serves the +// readiness endpoint once startMarker exists and closes its listener once +// stopMarker does, as a daemon told to stop by its control CLI would. It +// returns the daemon's port. +func commandDaemon(t *testing.T, startMarker, stopMarker string) int { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if fileExists(startMarker) && !fileExists(stopMarker) { + w.WriteHeader(http.StatusOK) + return + } + w.WriteHeader(http.StatusServiceUnavailable) + })) + t.Cleanup(srv.Close) + stopWatch := make(chan struct{}) + t.Cleanup(func() { close(stopWatch) }) + go func() { + for { + select { + case <-stopWatch: + return + case <-time.After(50 * time.Millisecond): + if fileExists(stopMarker) { + srv.Close() + return + } + } + } + }() + u, err := url.Parse(srv.URL) + require.NoError(t, err, "parse daemon URL") + port, err := strconv.Atoi(u.Port()) + require.NoError(t, err, "parse daemon port") + return port +} + +// TestStopAll_RetriesAFailedStopCommand checks that a later StopAll caller +// retries an engine whose stop command failed during the sweep, so a stop +// command that errors once does not leave the engine running after quit. +func TestStopAll_RetriesAFailedStopCommand(t *testing.T) { + dir := t.TempDir() + startMarker := filepath.Join(dir, "started") + stopAttempt := filepath.Join(dir, "stop-attempted") + stopMarker := filepath.Join(dir, "stopped") + port := commandDaemon(t, startMarker, stopMarker) + ex := newTestExecutor(t, &Manifest{ + Engine: "daemon", DisplayName: "Daemon Engine", ManifestVersion: 1, + Platforms: map[string]Platform{ + hostKey(): { + Detect: []string{fakeEngineBin}, + Runtime: Runtime{ + Mode: "command", + Port: port, + Start: [][]string{{fakeEngineBin, "touch", startMarker}}, + Stop: &StopSpec{Cmd: []string{fakeEngineBin, "failoncethentouch", stopAttempt, stopMarker}}, + Ready: &Probe{HTTP: "http://127.0.0.1:{port}/", Status: http.StatusOK, TimeoutS: 10}, + }, + }, + }, + }) + require.NoError(t, ex.Start(context.Background(), "daemon")) + ex.StopAll() + st, err := ex.Status("daemon") + require.NoError(t, err) + require.True(t, st.Running, "the failed stop must leave the engine running after the sweep") + + ex.StopAll() + + assert.FileExists(t, stopMarker, "the later StopAll must run the stop command again") + st, err = ex.Status("daemon") + require.NoError(t, err) + assert.False(t, st.Running, "the retried stop must leave the engine stopped") +} diff --git a/services/nvpair-engine-manager/testdata/fakeengine/main.go b/services/nvpair-engine-manager/testdata/fakeengine/main.go index 76626c20b..8100acffb 100644 --- a/services/nvpair-engine-manager/testdata/fakeengine/main.go +++ b/services/nvpair-engine-manager/testdata/fakeengine/main.go @@ -97,6 +97,16 @@ func main() { } fmt.Fprintf(os.Stderr, "Error: Failed to resolve artifact %q: The artifact does not exist or you do not have permission to read it\n", strings.ToLower(arg)) os.Exit(1) + case "failoncethentouch": // fail once, then write a marker for stop-retry tests + if len(os.Args) < 4 { + os.Exit(2) + } + if _, err := os.Stat(os.Args[2]); err != nil { + _ = os.WriteFile(os.Args[2], []byte("1"), 0o644) + os.Exit(1) + } + _ = os.WriteFile(os.Args[3], []byte("ok"), 0o644) + return case "noserve": // run but never bind — used to test readiness timeout time.Sleep(time.Hour) return diff --git a/services/nvpair-ui-broker/README.md b/services/nvpair-ui-broker/README.md index c94c943f6..58a359eed 100644 --- a/services/nvpair-ui-broker/README.md +++ b/services/nvpair-ui-broker/README.md @@ -703,6 +703,10 @@ Opt into / out of the `engine:` stream (off by default). Acks `{ subscrib Any other `engine:*` request is forwarded to `nvpair-engine-manager` verbatim and its response relayed straight back. This covers the whole engine control plane: `engine:get-installed`, `engine:describe`, `engine:status`, `engine:install`, `engine:uninstall`, `engine:start`, `engine:stop`, `engine:restart`, `engine:set-port`, `engine:action`, `engine:logs`, `engine:errors`, `engine:models`. (`engine:set-port` persists the engine's server port as a manifest override that survives a restart — see the `nvpair-engine-manager` README.) Lifecycle ops run for minutes (reporting progress via the `engine:install-progress` / `engine:state-changed` push events), so the relay imposes **no broker-side timeout** — fire the request and watch the event stream for the outcome. Error `-32000 "engine-manager not available"` when no engine-manager is supervised. +The broker's prepare-shutdown request and subsequent stdin close join the same +engine-manager shutdown sweep; the EOF backstop remains, without re-probing +foreign listeners a second time. Failed actionable stops can still be retried. + Engine-manager's internal `engine:settings-request{id, method}` is handled separately from subscriptions. Only `settings/get-engine-allow-lan-bind` is forwarded to node-settings, with a two-second bound, and the correlated From b3dda212049878df3d191fc09539a864718be186 Mon Sep 17 00:00:00 2001 From: Jared Ping <13906915+JaredP94@users.noreply.github.com> Date: Tue, 3 Nov 2026 00:33:20 +0200 Subject: [PATCH 5/5] fix(engine-manager): clean up canceled detached command starts The detached-daemon test observed a marker written before the start CLI exited. Shutdown could cancel the CLI before cmd.Wait returned, leaving the successful-command flag unset and skipping the official stop command despite a live detached daemon. Preserve command exit errors and clean up commands that actually executed, including canceled or nonzero exits, while excluding spawn failures. Synchronize the original readiness-phase test on a real readiness request and add a deterministic cancellation-before-CLI-exit regression. Also wait for the crash report itself in the separate watcher test instead of racing its earlier stopped-state publication. The detached cases pass -race -count=20 in simultaneous processes with broker relay load. Existing unreleased engine-manager 1.4.0 covers this fix relative to v0.8.0. Signed-off-by: Jared Ping <13906915+JaredP94@users.noreply.github.com> --- services/nvpair-engine-manager/README.md | 8 ++++ services/nvpair-engine-manager/install.go | 2 +- services/nvpair-engine-manager/lifecycle.go | 8 ++++ .../nvpair-engine-manager/remediation_test.go | 4 +- .../nvpair-engine-manager/shutdown_test.go | 44 ++++++++++++++++++- services/nvpair-engine-manager/status_test.go | 8 +++- .../testdata/fakeengine/main.go | 7 +++ 7 files changed, 77 insertions(+), 4 deletions(-) diff --git a/services/nvpair-engine-manager/README.md b/services/nvpair-engine-manager/README.md index ab2fe48a0..eea0d9533 100644 --- a/services/nvpair-engine-manager/README.md +++ b/services/nvpair-engine-manager/README.md @@ -242,6 +242,14 @@ a successful command must leave the tool detectable. | `-32011` | `unsupported-platform`: the node is not `darwin/arm64`. | | `-32012` | `install-failed`: the install command failed or macmon was not detected afterwards. Command failures include only the last 20 stderr lines, capped to the last 4096 bytes; stdout is discarded. | +### Startup cancellation + +A command-mode control CLI can launch its daemon before the CLI exits. Failed +or canceled starts run the bounded official stop command if a CLI actually ran, +including nonzero exits, before releasing the lifecycle lock. A command that +could not spawn does not authorize cleanup. Shutdown cancellation therefore +cannot skip cleanup merely because readiness had not begun. + ### Shutdown sweep `engine:prepare-shutdown` and the EOF/transport-exit path join one concurrent diff --git a/services/nvpair-engine-manager/install.go b/services/nvpair-engine-manager/install.go index 7f1e910a7..543e13d74 100644 --- a/services/nvpair-engine-manager/install.go +++ b/services/nvpair-engine-manager/install.go @@ -414,7 +414,7 @@ func (e *Executor) runCommand(ctx context.Context, argv []string) error { configureSysProcAttr(cmd) // hide the console window on Windows out, err := cmd.CombinedOutput() if err != nil { - return fmt.Errorf("%v: %s", err, strings.TrimSpace(string(out))) + return fmt.Errorf("%w: %s", err, strings.TrimSpace(string(out))) } return nil } diff --git a/services/nvpair-engine-manager/lifecycle.go b/services/nvpair-engine-manager/lifecycle.go index 0c2f829d9..3e1eb08fd 100644 --- a/services/nvpair-engine-manager/lifecycle.go +++ b/services/nvpair-engine-manager/lifecycle.go @@ -12,6 +12,7 @@ import ( "net/http" "net/url" "os" + "os/exec" "runtime" "strconv" "strings" @@ -305,6 +306,13 @@ func (e *Executor) bringUpCommand(ctx context.Context, st *engineState, engine s argv[i] = expandPath(argv[i]) } if err := e.runCommand(ctx, argv); err != nil { + // The CLI may have detached a daemon before exiting nonzero or + // being canceled. An ExitError proves it actually ran; clean up + // its possible side effects, but not a command that failed to spawn. + var exitErr *exec.ExitError + if errors.As(err, &exitErr) { + started = true + } werr := fmt.Errorf("start command failed: %w", err) e.reportStartFailedUnlessShuttingDown(ctx, engine, werr) return cleanup(werr) diff --git a/services/nvpair-engine-manager/remediation_test.go b/services/nvpair-engine-manager/remediation_test.go index 5e1ee3b30..f3ee0a90e 100644 --- a/services/nvpair-engine-manager/remediation_test.go +++ b/services/nvpair-engine-manager/remediation_test.go @@ -349,7 +349,9 @@ func TestUnexpectedExitReported(t *testing.T) { st, _ := ex.Status("fake") // The /exit route makes the fake engine os.Exit(1) — a crash. _, _ = http.Get(fmt.Sprintf("http://127.0.0.1:%d/exit", st.Port)) - waitFor(t, 6*time.Second, func() bool { s, _ := ex.Status("fake"); return !s.Running }) + // The watcher publishes stopped state before reporting the crash. Wait + // for the behavior under test, not the earlier state mutation. + waitFor(t, 6*time.Second, func() bool { return hasErr(ex.Errors(), exitedID("fake")) }) if !hasErr(ex.Errors(), exitedID("fake")) { t.Fatalf("expected an 'exited' error after a crash, got %+v", ex.Errors()) } diff --git a/services/nvpair-engine-manager/shutdown_test.go b/services/nvpair-engine-manager/shutdown_test.go index 93234bd72..fa98d1b66 100644 --- a/services/nvpair-engine-manager/shutdown_test.go +++ b/services/nvpair-engine-manager/shutdown_test.go @@ -13,6 +13,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "testing" "time" ) @@ -108,6 +109,16 @@ func TestStopAllCancelsCommandStartWithoutError(t *testing.T) { } func TestStopAllStopsDetachedCommandDaemonBeforeReadiness(t *testing.T) { + testStopAllDetachedCommandDaemon(t, "touch") +} + +// The CLI can launch its daemon before returning. Cancellation in that window +// must run the stop CLI too, even though the start command exits unsuccessfully. +func TestStopAllStopsDetachedCommandDaemonBeforeCLIExit(t *testing.T) { + testStopAllDetachedCommandDaemon(t, "touchwait") +} + +func testStopAllDetachedCommandDaemon(t *testing.T, command string) { dir := t.TempDir() startMarker := filepath.Join(dir, "started") stopMarker := filepath.Join(dir, "stopped") @@ -121,7 +132,7 @@ func TestStopAllStopsDetachedCommandDaemonBeforeReadiness(t *testing.T) { platform.Runtime.Mode = "command" platform.Runtime.Bin = "" platform.Runtime.Port = port - platform.Runtime.Start = [][]string{{fakeEngineBin, "touch", startMarker}} + platform.Runtime.Start = [][]string{{fakeEngineBin, command, startMarker}} platform.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineBin, "touch", stopMarker}, GraceS: 1} platform.Runtime.Ready = &Probe{ HTTP: "http://127.0.0.1:{port}/", @@ -135,7 +146,10 @@ func TestStopAllStopsDetachedCommandDaemonBeforeReadiness(t *testing.T) { // Model a control CLI that launches a detached daemon and returns before // that daemon becomes ready. The daemon deliberately answers 503 until its // stop CLI runs, which keeps Start inside waitReady. + readinessStarted := make(chan struct{}) + var readyOnce sync.Once daemon := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + readyOnce.Do(func() { close(readinessStarted) }) w.WriteHeader(http.StatusServiceUnavailable) })} t.Cleanup(func() { _ = daemon.Close() }) @@ -184,6 +198,15 @@ func TestStopAllStopsDetachedCommandDaemonBeforeReadiness(t *testing.T) { if !portServing(port) { t.Fatal("detached command daemon did not start serving") } + if command == "touch" { + // A marker written by the CLI does not prove cmd.Wait has returned. + // Observe readiness I/O to synchronize with the intended startup phase. + select { + case <-readinessStarted: + case <-time.After(5 * time.Second): + t.Fatal("start did not enter readiness wait") + } + } ex.StopAll() @@ -203,6 +226,25 @@ func TestStopAllStopsDetachedCommandDaemonBeforeReadiness(t *testing.T) { } } +func TestFailedCommandSpawnDoesNotRunStopCLI(t *testing.T) { + marker := filepath.Join(t.TempDir(), "stopped") + manifest := testEngineManifest(fakeEngineBin) + p := manifest.Platforms[hostKey()] + p.Runtime.Mode = "command" + p.Runtime.Bin = "" + p.Runtime.Start = [][]string{{filepath.Join(t.TempDir(), "nonexistent")}} + p.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineBin, "touch", marker}} + p.Runtime.Ready, p.Runtime.Health = nil, nil + manifest.Platforms[hostKey()] = p + ex := newTestExecutor(t, manifest) + if err := ex.Start(context.Background(), "fake"); err == nil { + t.Fatal("expected spawn failure") + } + if fileExists(marker) { + t.Fatal("spawn failure invoked stop CLI despite no started process") + } +} + // TestE2EStdinCloseStopsEngine drives the real engine-manager binary and stops // it the way the broker now does on shutdown: by closing stdin (EOF), with no // prepare-shutdown or shutdown RPC. The EOF path must run StopAll and take the diff --git a/services/nvpair-engine-manager/status_test.go b/services/nvpair-engine-manager/status_test.go index 748f3cb24..fb1fd0b96 100644 --- a/services/nvpair-engine-manager/status_test.go +++ b/services/nvpair-engine-manager/status_test.go @@ -67,7 +67,13 @@ func TestStatus_DoesNotReadoptAfterStopDuringProbe(t *testing.T) { status, err := ex.Status("fake") finished <- statusResult{status, err} }() - awaitProbeStart(t, fixture.probeStarted) + select { + case <-fixture.probeStarted: + case got := <-finished: + t.Fatalf("passive status returned before readiness request: %+v err=%v", got.status, got.err) + case <-time.After(2 * time.Second): + t.Fatal("readiness did not start and status still pending") + } // Bound the wait for Stop so a lock regression fails promptly. stopped := make(chan error, 1) diff --git a/services/nvpair-engine-manager/testdata/fakeengine/main.go b/services/nvpair-engine-manager/testdata/fakeengine/main.go index 8100acffb..94370d061 100644 --- a/services/nvpair-engine-manager/testdata/fakeengine/main.go +++ b/services/nvpair-engine-manager/testdata/fakeengine/main.go @@ -80,6 +80,13 @@ func main() { _ = os.WriteFile(os.Args[2], []byte("ok"), 0o644) } return + case "touchwait": // publish daemon side effects before the control CLI exits + if len(os.Args) < 3 { + os.Exit(2) + } + _ = os.WriteFile(os.Args[2], []byte("ok"), 0o644) + time.Sleep(time.Hour) + return case "echo": // print args to stdout so a cmd-action can capture output fmt.Println(strings.Join(os.Args[2:], " ")) return