diff --git a/docs/architecture.mdx b/docs/architecture.mdx index 3cd92d2b8..446d20c22 100644 --- a/docs/architecture.mdx +++ b/docs/architecture.mdx @@ -816,7 +816,11 @@ memberships and retire removed or re-addressed senders. The responder reuses its senders for announcements and replies too. Shutdown closes the sockets after the responder's goodbye. A standalone `Browser.Poll` owns a one-shot transport; use `Run` for periodic discovery. The scan cadence and per-interface send-failure -feedback used by address selection are unchanged. +feedback used by address selection are unchanged. Partial browser receive-setup +failures retry only the failed family binds or interface memberships at scan +boundaries, with exponential backoff from 1 second to a 60-second cap. Successful +sockets and memberships stay in place; a changed network snapshot rebuilds setup +immediately rather than waiting for that backoff. This reduces measured socket and interface-query churn. It does **not** establish that discovery caused the macOS kernel panics or WLAN failures reported in diff --git a/services/nvpair-engine-manager/MANIFEST.md b/services/nvpair-engine-manager/MANIFEST.md index 40d9b1258..ce7169422 100644 --- a/services/nvpair-engine-manager/MANIFEST.md +++ b/services/nvpair-engine-manager/MANIFEST.md @@ -3,7 +3,34 @@ SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All SPDX-License-Identifier: Apache-2.0 --> -# Editable launch manifest bindings +# Engine manifest capabilities + +## Command-mode shutdown confirmation + +`runtime.stop.wait_for_exit` is an optional boolean (default `false`). It +requires `runtime.mode: "command"`, a nonempty `runtime.stop.cmd`, and a +`runtime.ready` probe; manifests enabling it without these are rejected. + +By default, command-mode shutdown confirms three consecutive connection +refusals on the readiness listener. A persistent host process, such as LM +Studio after `lms server stop`, need not exit; its server can be restarted. +Set `wait_for_exit: true` only when the app must fully exit before restarting. +Only bundled oMLX enables it: + +```json +"stop": { "cmd": ["{cli}", "stop"], "wait_for_exit": true } +``` + +With this capability, PAIR also waits for the serving PID observed before the +stop command to exit, when the host can resolve it. If PID lookup is unavailable, +listener closure remains the available evidence. The stop command is bounded by +`max(grace_s, 5s)` and confirmation by `grace_s` (default 5s), without extending +the deadline. Timeout retains running state and reports an error. The state +mutex is never held across lookup, command execution, or confirmation; the +lifecycle operation lock still serializes stop/start. Failed-start cleanup uses +the same confirmation policy. + +## Editable launch manifest bindings The engine-manager runner remains manifest-driven. `runtime.editable_launch` identifies the fixed startup arguments and PAIR-owned networking controls; it diff --git a/services/nvpair-engine-manager/README.md b/services/nvpair-engine-manager/README.md index 39b6ccb5c..207b1467b 100644 --- a/services/nvpair-engine-manager/README.md +++ b/services/nvpair-engine-manager/README.md @@ -600,9 +600,11 @@ terminate is the only signal that actually stops them. Command-mode stop (including oMLX) gives the official stop CLI `max(grace_s, 5s)` to return, then up to `grace_s` (default 5s) for shutdown. -It confirms three consecutive connection refusals **and**, when the host can -resolve it, exit of the process observed on the port before stop. Closing the -port alone does not mean app teardown is complete. The state mutex is never +It confirms three consecutive connection refusals. Only manifests opting into +`runtime.stop.wait_for_exit` (bundled oMLX) also require exit of the process +observed on the port before stop, when the host can resolve it. LM Studio stops +only its HTTP server; its persistent host need not exit for stop/restart. +See [MANIFEST.md](MANIFEST.md#command-mode-shutdown-confirmation). The state mutex is never held across these waits; lifecycle operations remain serialized so an immediate start cannot overtake stop. A timeout leaves the engine running, not falsely stopped, and is returned to the caller. diff --git a/services/nvpair-engine-manager/lifecycle.go b/services/nvpair-engine-manager/lifecycle.go index d56c9ea9d..ca58de9df 100644 --- a/services/nvpair-engine-manager/lifecycle.go +++ b/services/nvpair-engine-manager/lifecycle.go @@ -561,12 +561,14 @@ func (e *Executor) runCommandStop(st *engineState, engine string, rt Runtime, po // Bound the stop CLI (e.g. `lms server stop`): the broker no longer // force-kills engine-manager on a timeout, so an unbounded stop command // would wedge StopAll and, in turn, the whole app shutdown. - // Snapshot the serving PID before the CLI can close its listener. The - // daemon can remain alive tearing down state after the port is released. + // Only engines declaring wait_for_exit need the serving PID to exit: + // some apps tear down after releasing the port, while others (LM Studio) + // intentionally keep their host process alive after stopping the server. + // Snapshot the PID before the CLI can close its listener. // No state mutex is held during lookup, command execution, or this wait; // opMu still serializes lifecycle operations, as it does for readiness. pid := 0 - if rt.Ready != nil { + if sp.WaitForExit && rt.Ready != nil { pid, _, _ = pidOnPort(port) // Never wait for this service itself to exit (e.g. a misconfigured // probe targeting engine-manager). Only an independent daemon counts. @@ -648,8 +650,9 @@ func (e *Executor) waitUnavailableContext(parent context.Context, p *Probe, port return e.waitDownContext(parent, p, port, 0, timeout) } -// waitCommandUnavailable requires both transport closure and exit of the PID -// observed before stop. If the host cannot resolve a PID, transport remains +// waitCommandUnavailable requires transport closure and, for engines opting +// into wait_for_exit, exit of the PID observed before stop. Otherwise pid is 0. +// If the host cannot resolve a PID, transport remains // the best available evidence; the deadline is never extended by polling. func (e *Executor) waitCommandUnavailable(p *Probe, port, pid int, timeout time.Duration) bool { return e.waitDownContext(context.Background(), p, port, pid, timeout) diff --git a/services/nvpair-engine-manager/manifests/omlx.json b/services/nvpair-engine-manager/manifests/omlx.json index e95e7b3b7..70a34a0e4 100644 --- a/services/nvpair-engine-manager/manifests/omlx.json +++ b/services/nvpair-engine-manager/manifests/omlx.json @@ -50,7 +50,7 @@ "port": 1235, "bind": "127.0.0.1", "start": [["{cli}", "start"]], - "stop": { "cmd": ["{cli}", "stop"] }, + "stop": { "cmd": ["{cli}", "stop"], "wait_for_exit": true }, "ready": { "http": "http://127.0.0.1:{port}/health", "status": 200, "timeout_s": 45 }, "health": { "http": "http://127.0.0.1:{port}/health", "status": 200, "interval_s": 10 } }, diff --git a/services/nvpair-engine-manager/registry.go b/services/nvpair-engine-manager/registry.go index 9bab2ac09..587b92978 100644 --- a/services/nvpair-engine-manager/registry.go +++ b/services/nvpair-engine-manager/registry.go @@ -238,6 +238,9 @@ type StopSpec struct { Signal string `json:"signal,omitempty"` // "term" (default) | "kill" Cmd []string `json:"cmd,omitempty"` // optional explicit stop command GraceS int `json:"grace_s,omitempty"` + // WaitForExit requires command-mode shutdown to confirm the serving PID + // exits as well as releasing the listener. Most apps only stop the server. + WaitForExit bool `json:"wait_for_exit,omitempty"` } // Action is a config-declared operation exposed over engine:action. @@ -778,6 +781,11 @@ func (p *Platform) validate(key string) error { } } } + if sp := p.Runtime.Stop; sp != nil && sp.WaitForExit { + if p.Runtime.modeOrDefault() != "command" || len(sp.Cmd) == 0 || p.Runtime.Ready == nil { + return fmt.Errorf("platform %q: runtime.stop.wait_for_exit requires command mode, stop.cmd, and runtime.ready", key) + } + } if err := validateProbe(key, "ready", p.Runtime.Ready); err != nil { return err } diff --git a/services/nvpair-engine-manager/registry_test.go b/services/nvpair-engine-manager/registry_test.go index db2624023..7e840cf2c 100644 --- a/services/nvpair-engine-manager/registry_test.go +++ b/services/nvpair-engine-manager/registry_test.go @@ -5,6 +5,7 @@ package main import ( "encoding/json" + "fmt" "os" "path/filepath" "slices" @@ -69,6 +70,40 @@ func TestValidateAcceptsCommandModeAndCmdAction(t *testing.T) { } } +func TestValidateStopWaitForExit(t *testing.T) { + for _, tc := range []struct { + name, mode, wait, cmd, ready string + wantErr bool + }{ + {"enabled command", "command", "true", `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, false}, + {"disabled command", "command", "false", `["cli","stop"]`, `null`, false}, + {"disabled process", "process", "false", `[]`, `null`, false}, + {"process mode", "process", "true", `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, true}, + {"missing stop command", "command", "true", `[]`, `{"tcp":"127.0.0.1:{port}"}`, true}, + {"missing readiness probe", "command", "true", `["cli","stop"]`, `null`, true}, + {"string instead of boolean", "command", `"true"`, `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, true}, + {"number instead of boolean", "command", "1", `["cli","stop"]`, `{"tcp":"127.0.0.1:{port}"}`, true}, + } { + t.Run(tc.name, func(t *testing.T) { + data := fmt.Sprintf(`{ + "engine":"test", "display_name":"Test", "manifest_version":1, + "platforms":{"linux/amd64":{"runtime":{ + "mode":%q, "bin":"cli", "start":[["cli","start"]], + "stop":{"cmd":%s,"wait_for_exit":%s}, "ready":%s + }}} + }`, tc.mode, tc.cmd, tc.wait, tc.ready) + var m Manifest + err := json.Unmarshal([]byte(data), &m) + if err == nil { + err = m.Validate() + } + if (err != nil) != tc.wantErr { + t.Fatalf("manifest validation error = %v, wantErr=%v", err, tc.wantErr) + } + }) + } +} + func TestValidateAcceptsUnpinnedFetch(t *testing.T) { m := validManifest() p := m.Platforms["linux/amd64"] @@ -384,6 +419,25 @@ func TestBundledManifestsMerge(t *testing.T) { } } +func TestBundledStopWaitForExit(t *testing.T) { + reg := NewRegistry() + if err := reg.LoadFS(bundledManifests, "manifests"); err != nil { + t.Fatal(err) + } + for _, engine := range []string{"omlx", "lmstudio", "ollama"} { + m, ok := reg.Get(engine) + if !ok { + t.Fatalf("missing bundled engine %s", engine) + } + for key, p := range m.Platforms { + wait := p.Runtime.Stop != nil && p.Runtime.Stop.WaitForExit + if wait != (engine == "omlx") { + t.Errorf("%s %s: wait_for_exit=%v, only oMLX requires app exit before restart", engine, key, wait) + } + } + } +} + // TestBundledOllamaReadinessBudget pins the finite startup allowance used by // every supported platform. Ollama does not serve /api/version until GPU // discovery completes, which can exceed the previous 30-second allowance. diff --git a/services/nvpair-engine-manager/shutdown_tail_test.go b/services/nvpair-engine-manager/shutdown_tail_test.go index c4a06325f..4afa64d02 100644 --- a/services/nvpair-engine-manager/shutdown_tail_test.go +++ b/services/nvpair-engine-manager/shutdown_tail_test.go @@ -4,6 +4,7 @@ package main import ( + "context" "errors" "os/exec" "path/filepath" @@ -12,6 +13,63 @@ import ( "time" ) +func TestCommandStopAndRestartWithPersistentHost(t *testing.T) { + marker := filepath.Join(t.TempDir(), "server") + port, err := freePort() + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(fakeEngineForTest(t), "persistent-host", marker, "127.0.0.1:"+strconv.Itoa(port)) + configureSysProcAttr(cmd) + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + go func() { _ = cmd.Wait(); close(done) }() + t.Cleanup(func() { killTestProcessGroup(cmd.Process.Pid); <-done }) + waitPortServing(t, port) + if pid, _, ok := pidOnPort(port); !ok || pid != cmd.Process.Pid { + t.Skip("host cannot resolve listener PID") + } + m := testEngineManifest(fakeEngineForTest(t)) + p := m.Platforms[hostKey()] + p.Runtime.Mode = "command" + p.Runtime.Port = port + p.Runtime.Start = [][]string{{fakeEngineForTest(t), "touch", marker + "-start"}} + p.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineForTest(t), "touch", marker + "-stop"}, GraceS: 1} + p.Runtime.Health = nil + m.Platforms[hostKey()] = p + ex := newTestExecutor(t, m) + st, _ := ex.state("fake") + st.running, st.port = true, port + + if err := ex.Stop("fake"); err != nil { + t.Fatalf("stop must succeed when the server closes but its host stays alive: %v", err) + } + if status, _ := ex.Status("fake"); status.Running || status.Healthy || portServing(port) { + t.Fatalf("stopped server must be reported down: %+v", status) + } + select { + case <-done: + t.Fatal("stop terminated the persistent host") + default: + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := ex.Start(ctx, "fake"); err != nil { + t.Fatalf("start after stop: %v", err) + } + if err := ex.Restart(ctx, "fake"); err != nil { + t.Fatalf("restart with a persistent host: %v", err) + } + if status, _ := ex.Status("fake"); !status.Running || !status.Healthy || !portServing(port) { + t.Fatalf("restarted server must be healthy: %+v", status) + } + if pid, _, ok := pidOnPort(port); !ok || pid != cmd.Process.Pid { + t.Fatalf("restart must reuse the persistent host, got pid=%d resolved=%v", pid, ok) + } +} + func TestCommandStopWaitsForProcessExitAfterPortCloses(t *testing.T) { testCommandStopTail(t, "900ms", 2, false) } @@ -43,7 +101,7 @@ func testCommandStopTail(t *testing.T, delay string, grace int, wantTimeout bool p := m.Platforms[hostKey()] p.Runtime.Mode = "command" p.Runtime.Port = port - p.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineForTest(t), "touch", marker}, GraceS: grace} + p.Runtime.Stop = &StopSpec{Cmd: []string{fakeEngineForTest(t), "touch", marker}, GraceS: grace, WaitForExit: true} p.Runtime.Health = nil m.Platforms[hostKey()] = p ex := newTestExecutor(t, m) diff --git a/services/nvpair-engine-manager/shutdown_test.go b/services/nvpair-engine-manager/shutdown_test.go index 2e215dc53..b393e6846 100644 --- a/services/nvpair-engine-manager/shutdown_test.go +++ b/services/nvpair-engine-manager/shutdown_test.go @@ -142,6 +142,8 @@ func testStopAllDetachedCommandDaemon(t *testing.T, command string) { platform.Runtime.Health = nil manifest.Platforms[hostKey()] = platform ex := newTestExecutor(t, manifest) + // This must run even if fixture setup/readiness fails before explicit StopAll. + t.Cleanup(ex.StopAll) // Model a control CLI that launches a detached daemon and returns before // that daemon becomes ready. The daemon deliberately answers 503 until its @@ -195,9 +197,7 @@ func testStopAllDetachedCommandDaemon(t *testing.T, command string) { if err := <-daemonStarted; err != nil { t.Fatalf("detached command daemon: %v", err) } - if !portServing(port) { - t.Fatal("detached command daemon did not start serving") - } + waitPortServing(t, port) 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. diff --git a/services/nvpair-engine-manager/status_test.go b/services/nvpair-engine-manager/status_test.go index 8831bc2a8..6460ff672 100644 --- a/services/nvpair-engine-manager/status_test.go +++ b/services/nvpair-engine-manager/status_test.go @@ -254,15 +254,41 @@ func TestStartDoesNotWaitForPassiveProbe(t *testing.T) { awaitProbeStart(t, fixture.probeStarted) port, err := freePort() require.NoError(t, err) + // Keep the operation observable even if startup finishes between polls. + // Persistence is the last step of StartWith, after spawning and readiness. + ex.desired.mu.Lock() + var persistOnce sync.Once + allowPersistence := func() { persistOnce.Do(ex.desired.mu.Unlock) } + t.Cleanup(func() { + allowPersistence() + fixture.releaseProbe() + ex.StopAll() + }) 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)) + // Only lifecycle entry has this short deadline, not real process startup. + require.Eventually(t, func() bool { + st.mu.Lock() + defer st.mu.Unlock() + return st.opDone != nil + }, 2*time.Second, time.Millisecond, "Start waited for a passive readiness probe") + select { + case <-finished: + t.Fatal("passive probe finished before lifecycle entry was observed") + default: + } fixture.releaseProbe() - got := awaitResult(t, finished) - require.NoError(t, got.err) - require.True(t, got.status.Running) - require.Equal(t, port, got.status.Port) + require.NoError(t, awaitResult(t, finished).err) + allowPersistence() + select { + case err := <-started: + require.NoError(t, err) + case <-time.After(15 * time.Second): + t.Fatal("engine startup did not finish after lifecycle entry") + } + status := ex.snapshot("fake", st) + require.True(t, status.Running) + require.Equal(t, port, status.Port) } type statusResult struct { diff --git a/services/nvpair-engine-manager/testdata/fakeengine/main.go b/services/nvpair-engine-manager/testdata/fakeengine/main.go index 379700e32..7d65e2603 100644 --- a/services/nvpair-engine-manager/testdata/fakeengine/main.go +++ b/services/nvpair-engine-manager/testdata/fakeengine/main.go @@ -206,6 +206,36 @@ func main() { os.Exit(1) } return + case "persistent-host": + // Model an app whose HTTP server can stop/restart without exiting. + var server *http.Server + start := func() { + ln, err := net.Listen("tcp", os.Args[3]) + if err != nil { + os.Exit(2) + } + server = &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })} + go func(s *http.Server) { _ = s.Serve(ln) }(server) + } + start() + for { + if _, err := os.Stat(os.Args[2] + "-stop"); err == nil { + if server != nil { + _ = server.Close() + server = nil + } + _ = os.Remove(os.Args[2] + "-stop") + } + if _, err := os.Stat(os.Args[2] + "-start"); err == nil { + if server == nil { + start() + } + _ = os.Remove(os.Args[2] + "-start") + } + time.Sleep(10 * time.Millisecond) + } case "shutdown-tail": // Close the listener before asynchronous process teardown completes. ln, err := net.Listen("tcp", os.Args[3]) diff --git a/services/nvpair-node-scanner/README.md b/services/nvpair-node-scanner/README.md index 5bb208868..0fb26e707 100644 --- a/services/nvpair-node-scanner/README.md +++ b/services/nvpair-node-scanner/README.md @@ -42,6 +42,11 @@ on interface-change notifications rather than re-enumerated on every five-second scan. The responder reuses per-interface senders for announcements and replies; removed or re-addressed interfaces retire their cached senders, including when all eligible interfaces disappear. Shutdown sends the goodbye before closing them. +If only part of browser receive setup succeeds, failed address-family binds and +interface memberships retry at scan boundaries with exponential backoff (1 second +up to 60 seconds), without replacing working sockets or rejoining working +memberships. A network snapshot change rebuilds setup immediately; a failed +established receiver is rebuilt on the next scan. Every scan still performs real per-interface writes, preserving the consecutive send-failure evidence that address selection consumes. Source binding preserves diff --git a/services/shared/discovery/network.go b/services/shared/discovery/network.go index 9fc10b1a7..5d76831d1 100644 --- a/services/shared/discovery/network.go +++ b/services/shared/discovery/network.go @@ -12,6 +12,7 @@ import ( "reflect" "strings" "sync" + "time" "nvpair-shared/mdns" "nvpair-shared/netmon" @@ -44,6 +45,7 @@ func newBrowserNetwork(ctx context.Context) *browserNetwork { func (n *browserNetwork) refresh(snapshot netmon.Snapshot) error { if n.resolver != nil && n.resolver.healthy() && reflect.DeepEqual(n.snapshot, snapshot) { + n.resolver.retrySetup(n.ctx) return nil } if n.resolver != nil { @@ -65,6 +67,7 @@ func (n *browserNetwork) prepare() error { if n.resolver == nil || !n.resolver.healthy() { return n.refresh(n.monitor.Snapshot()) } + n.resolver.retrySetup(n.ctx) return nil } } @@ -100,10 +103,50 @@ type resolver struct { readers sync.WaitGroup failed chan struct{} failureOnce sync.Once + + // Setup retries run only on the serialized scan path. Readers never access + // these fields; successful sockets and memberships survive partial failures. + families []receiveFamily + open func(context.Context, string, string) (*receiveSocket, error) + now func() time.Time + retryAt time.Time + retryDelay time.Duration +} + +type receiveFamily struct { + network, address, group string + socket *receiveSocket + pending []net.Interface +} + +// receiveSocket keeps the platform multicast operations alongside their socket. +type receiveSocket struct { + net.PacketConn + join func(*net.Interface, net.Addr) error + v6 *ipv6.PacketConn +} + +func openReceiveSocket(ctx context.Context, network, address string) (*receiveSocket, error) { + conn, err := mdns.ListenPacket(ctx, network, address) + if err != nil { + return nil, err + } + socket := &receiveSocket{PacketConn: conn} + if network == "udp4" { + socket.join = ipv4.NewPacketConn(conn).JoinGroup + } else { + socket.v6 = ipv6.NewPacketConn(conn) + socket.join = socket.v6.JoinGroup + } + return socket, nil } func newResolver(ctx context.Context, snapshot netmon.Snapshot, packets chan *dns.Msg) (*resolver, error) { - r := &resolver{packets: packets, failed: make(chan struct{})} + return newResolverWithReceiver(ctx, snapshot, packets, openReceiveSocket, time.Now) +} + +func newResolverWithReceiver(ctx context.Context, snapshot netmon.Snapshot, packets chan *dns.Msg, open func(context.Context, string, string) (*receiveSocket, error), now func() time.Time) (*resolver, error) { + r := &resolver{packets: packets, failed: make(chan struct{}), open: open, now: now} for _, ifi := range snapshot.Interfaces { if ifi.Flags&net.FlagUp != 0 && ifi.Flags&net.FlagMulticast != 0 { r.interfaces = append(r.interfaces, ifi) @@ -112,54 +155,92 @@ func newResolver(ctx context.Context, snapshot netmon.Snapshot, packets chan *dn if len(r.interfaces) == 0 { return nil, errors.New("no multicast interfaces") } - for _, family := range []struct{ network, address, group string }{{"udp4", "224.0.0.251:5353", "224.0.0.251"}, {"udp6", "[ff02::fb]:5353", "ff02::fb"}} { - conn, err := mdns.ListenPacket(ctx, family.network, family.address) - if err != nil { - slog.Debug("mdns: receive socket unavailable", "network", family.network, "err", err) + r.families = []receiveFamily{ + {network: "udp4", address: "224.0.0.251:5353", group: "224.0.0.251", pending: r.interfaces}, + {network: "udp6", address: "[ff02::fb]:5353", group: "ff02::fb", pending: r.interfaces}, + } + r.retrySetup(ctx) + if len(r.conns) == 0 { + return nil, errors.New("failed to join any mDNS receive group") + } + return r, nil +} + +// Retry only failed binds/memberships with 1s -> 60s capped backoff, at scan +// boundaries. An unavailable family must not churn the working receive path. +func (r *resolver) retrySetup(ctx context.Context) { + if ctx.Err() != nil || r.now().Before(r.retryAt) { + return + } + incomplete := false + for i := range r.families { + family := &r.families[i] + if len(family.pending) == 0 { continue } - group := &net.UDPAddr{IP: net.ParseIP(family.group)} - var join func(*net.Interface, net.Addr) error - var pc6 *ipv6.PacketConn - if family.network == "udp4" { - join = ipv4.NewPacketConn(conn).JoinGroup - } else { - pc6 = ipv6.NewPacketConn(conn) - join = pc6.JoinGroup + fresh := family.socket == nil + if fresh { + var err error + family.socket, err = r.open(ctx, family.network, family.address) + if err != nil { + slog.Debug("mdns: receive socket unavailable", "network", family.network, "err", err) + incomplete = true + continue + } } + group := &net.UDPAddr{IP: net.ParseIP(family.group)} + var pending []net.Interface joined := 0 - for _, ifi := range r.interfaces { - if err := join(&ifi, group); err == nil { + for _, ifi := range family.pending { + if err := family.socket.join(&ifi, group); err != nil { + pending = append(pending, ifi) + } else { joined++ } } - if joined == 0 { - _ = conn.Close() - continue + family.pending = pending + incomplete = incomplete || len(pending) > 0 + if fresh { + if joined == 0 { + _ = family.socket.Close() + family.socket = nil + continue + } + r.startReceiver(family.socket) } - r.conns = append(r.conns, conn) - if pc6 != nil { - r.v6 = pc6 - _ = pc6.SetMulticastHopLimit(255) + } + if incomplete { + if r.retryDelay == 0 { + r.retryDelay = time.Second + } else { + r.retryDelay = min(2*r.retryDelay, time.Minute) } - r.readers.Add(1) - go func() { - defer r.readers.Done() - buf := make([]byte, 65536) - for { - n, from, err := conn.ReadFrom(buf) - if err != nil { - r.failureOnce.Do(func() { close(r.failed) }) - return - } - r.receive(buf[:n], 0, from) - } - }() + r.retryAt = r.now().Add(r.retryDelay) + } else { + r.retryDelay = 0 + r.retryAt = time.Time{} } - if len(r.conns) == 0 { - return nil, errors.New("failed to join any mDNS receive group") +} + +func (r *resolver) startReceiver(conn *receiveSocket) { + r.conns = append(r.conns, conn) + if conn.v6 != nil { + r.v6 = conn.v6 + _ = conn.v6.SetMulticastHopLimit(255) } - return r, nil + r.readers.Add(1) + go func() { + defer r.readers.Done() + buf := make([]byte, 65536) + for { + n, from, err := conn.ReadFrom(buf) + if err != nil { + r.failureOnce.Do(func() { close(r.failed) }) + return + } + r.receive(buf[:n], 0, from) + } + }() } func (n *browserNetwork) receive(buf []byte, _ int, _ net.Addr) { diff --git a/services/shared/discovery/receiver_retry_test.go b/services/shared/discovery/receiver_retry_test.go new file mode 100644 index 000000000..492854105 --- /dev/null +++ b/services/shared/discovery/receiver_retry_test.go @@ -0,0 +1,133 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package discovery + +import ( + "context" + "errors" + "net" + "testing" + "time" + + "github.com/miekg/dns" + "nvpair-shared/netmon" +) + +func TestPartialReceiverSetupRetriesWithoutNetworkChanges(t *testing.T) { + for _, family := range []string{"udp4", "udp6"} { + for _, failure := range []string{"bind", "one-membership", "all-memberships"} { + t.Run(family+"/"+failure, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + now := time.Unix(1000, 0) + broken := true + opens := map[string]int{} + joins := map[string]map[int]int{"udp4": {}, "udp6": {}} + var sockets []net.PacketConn + latest := map[string]net.PacketConn{} + open := func(_ context.Context, network, _ string) (*receiveSocket, error) { + opens[network]++ + if broken && network == family && failure == "bind" { + return nil, errors.New("temporary bind failure") + } + conn, err := net.ListenPacket("udp4", "127.0.0.1:0") + if err != nil { + return nil, err + } + sockets = append(sockets, conn) + latest[network] = conn + return &receiveSocket{PacketConn: conn, join: func(ifi *net.Interface, _ net.Addr) error { + joins[network][ifi.Index]++ + if broken && network == family && (failure == "all-memberships" || failure == "one-membership" && ifi.Index == 2) { + return errors.New("temporary join failure") + } + return nil + }}, nil + } + snapshot := netmon.Snapshot{Interfaces: map[int]net.Interface{ + 1: {Index: 1, Name: "test1", Flags: net.FlagUp | net.FlagMulticast}, + 2: {Index: 2, Name: "test2", Flags: net.FlagUp | net.FlagMulticast}, + }} + packets := make(chan *dns.Msg, 1) + r, err := newResolverWithReceiver(ctx, snapshot, packets, open, func() time.Time { return now }) + if err != nil { + t.Fatal(err) + } + n := &browserNetwork{ctx: ctx, cancel: cancel, resolver: r, snapshot: snapshot} + defer func() { + n.close() + for _, conn := range sockets { + if err := conn.SetReadDeadline(time.Now()); !errors.Is(err, net.ErrClosed) { + t.Errorf("receiver leaked: %v", err) + } + } + }() + attempts := func() int { + if failure == "bind" { + return opens[family] + } + return joins[family][2] + } + // The scan boundary drives retries, not network events or sleeps. + for i, delay := range []time.Duration{1, 2, 4, 8, 16, 32, 60, 60} { + now = now.Add(delay*time.Second - time.Nanosecond) + if err := n.prepare(); err != nil { + t.Fatal(err) + } + if got := attempts(); got != i+1 { + t.Fatalf("retried before backoff elapsed: attempts=%d, want %d", got, i+1) + } + now = now.Add(time.Nanosecond) + if i == 7 { + broken = false + } + if err := n.prepare(); err != nil { + t.Fatal(err) + } + if got := attempts(); got != i+2 { + t.Fatalf("failed setup not retried after backoff: attempts=%d, want %d", got, i+2) + } + } + if n.resolver != r || len(r.conns) != 2 { + t.Fatalf("healthy resolver replaced or missing recovered family: connections=%d", len(r.conns)) + } + for network := range opens { + if network != family || failure == "one-membership" { + if opens[network] != 1 || joins[network][1] != 1 { + t.Errorf("successful socket/membership retried: %s opens=%d joins=%v", network, opens[network], joins[network]) + } + } + } + now = now.Add(time.Hour) + if err := n.prepare(); err != nil { + t.Fatal(err) + } + if got := attempts(); got != 9 { + t.Fatalf("setup retried after recovery: %d attempts", got) + } + // A recovered family must actually start its packet reader. + writer, err := net.Dial("udp4", latest[family].LocalAddr().String()) + if err != nil { + t.Fatal(err) + } + defer writer.Close() + buf, err := (&dns.Msg{MsgHdr: dns.MsgHdr{Response: true, Id: 42}}).Pack() + if err != nil { + t.Fatal(err) + } + if _, err := writer.Write(buf); err != nil { + t.Fatal(err) + } + select { + case msg := <-packets: + if msg.Id != 42 { + t.Fatalf("unexpected packet: %v", msg) + } + case <-time.After(time.Second): + t.Fatal("recovered receiver did not deliver a packet") + } + }) + } + } +}