From 9213215c65ab2776dd4ee6f16b6edc07ae735169 Mon Sep 17 00:00:00 2001 From: Jared Ping <13906915+JaredP94@users.noreply.github.com> Date: Fri, 6 Nov 2026 12:12:28 +0200 Subject: [PATCH 1/2] fix(cluster): retain authenticated peer address candidates across roaming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A paired machine can change networks while its member record retains only the old LAN endpoint. Learn a bounded, expiring set of endpoints from pinned roster connections and each authenticated peer’s self-report, retain the pairing-time fallback, and budget unproven dials across concurrent reconciles and restarts. Every application connection still verifies the exact pin and refuses redirects. Keep address hints local to members.json rather than extending the JSON-RPC projection. Serialize learning with admission removal, ignore discovery and third-party address claims, and update cluster-derived manual-node entries without changing saved entries. Add unit regressions and an isolated real-worker roaming/restart/removal test for CI. Bump cluster-manager for the additive roster field and manual-nodes for the endpoint-refresh fix. Signed-off-by: Jared Ping <13906915+JaredP94@users.noreply.github.com> --- docs/architecture.mdx | 19 ++ services/nvpair-cluster-manager/README.md | 42 +++ services/nvpair-cluster-manager/addresses.go | 283 ++++++++++++++++++ .../nvpair-cluster-manager/addresses_test.go | 151 ++++++++++ .../nvpair-cluster-manager/connreuse_test.go | 2 + services/nvpair-cluster-manager/httpserver.go | 13 +- services/nvpair-cluster-manager/manager.go | 14 +- services/nvpair-cluster-manager/membership.go | 76 +---- .../membership_store.go | 15 +- services/nvpair-cluster-manager/mtls.go | 33 +- .../nvpair-cluster-manager/peer_addr_test.go | 133 ++------ .../nvpair-cluster-manager/reconnect_test.go | 67 +---- services/nvpair-cluster-manager/remove.go | 6 +- .../roaming_order_test.go | 20 ++ .../roaming_persistence_test.go | 35 +++ .../roaming_redirect_test.go | 33 ++ .../nvpair-cluster-manager/roaming_test.go | 188 ++++++++++++ services/nvpair-cluster-manager/roster.go | 40 +-- .../nvpair-cluster-manager/roster_http.go | 179 ++++------- services/nvpair-manual-nodes/README.md | 5 + services/nvpair-manual-nodes/manager.go | 6 +- services/nvpair-manual-nodes/roaming_test.go | 63 ++++ services/tests/peer_roaming_test.go | 105 +++++++ services/versions.json | 4 +- 24 files changed, 1109 insertions(+), 423 deletions(-) create mode 100644 services/nvpair-cluster-manager/addresses.go create mode 100644 services/nvpair-cluster-manager/addresses_test.go create mode 100644 services/nvpair-cluster-manager/roaming_order_test.go create mode 100644 services/nvpair-cluster-manager/roaming_persistence_test.go create mode 100644 services/nvpair-cluster-manager/roaming_redirect_test.go create mode 100644 services/nvpair-cluster-manager/roaming_test.go create mode 100644 services/nvpair-manual-nodes/roaming_test.go create mode 100644 services/tests/peer_roaming_test.go diff --git a/docs/architecture.mdx b/docs/architecture.mdx index dd24eb23d..3cd92d2b8 100644 --- a/docs/architecture.mdx +++ b/docs/architecture.mdx @@ -254,6 +254,25 @@ and its broker-owned service metadata over pinned cluster mTLS on the same port. The public preview still exposes hostname, hardware inventory, and utilization to devices on the subnet; do not run PAIR where that disclosure is unacceptable. +### Authenticated Peer Address Roaming + +Cluster-manager keeps at most eight local address hints per member in +`cluster/members.json` (0600), ordered by authenticated success, retaining the +pairing-time endpoint as a fallback. Only a pinned roster connection's source +address, that peer's own `selfAddresses` report, or its successful pinned +response can teach that member's record. Discovery, previews, names and relayed +addresses cannot teach it. Learned hints expire after 30 days without success; +new/unproven endpoints are dial-budgeted to one per member per 30 seconds, +including across restarts. IPv6 scopes are local to the observed connection. + +An address is never an identity: every connection checks the exact current +certificate pin and redirects are refused. Removal clears the member's hints, +including against an in-flight read. Saved manual addresses retain user +precedence; cluster-derived entries follow the authenticated reached address in +the existing node detail. If both peers lose every known path, an explicit +reachable rendezvous is still required. See +[cluster-manager](../services/nvpair-cluster-manager/README.md#peer-address-roaming). + ### Two Personalities on One Port The proxies and `nvpair-cluster-manager` serve two different things on a single diff --git a/services/nvpair-cluster-manager/README.md b/services/nvpair-cluster-manager/README.md index f360a5f4c..eb5315337 100644 --- a/services/nvpair-cluster-manager/README.md +++ b/services/nvpair-cluster-manager/README.md @@ -363,6 +363,48 @@ cleaning or renaming it, and without touching disk or announcing a change. Broker-supplied address always wins, so manual-only nodes are still invitable and discovery-less networks still work. +## Peer address roaming + +`POST /v1/cluster/roster` adds optional `selfAddresses: string[]`: numeric +`host:port` endpoints for the **authenticated sender itself**, including routed +overlay addresses from the local interface publisher. Older peers may omit it. +The receiver learns only from that pinned connection's source host (keeping the +known listening port), the sender's self-report, or a successful pinned response. +Relayed roster addresses, names, mDNS and public previews never update this set. +Identity always comes from the exact certificate pin, not an address. + +Local-only `pairingAddress`, `addressCandidates` (endpoint, first learned time, +last authenticated success, last attempt), and `nextAddressProbe` are persisted +atomically inside `cluster/members.json` with mode **0600**. These fields are not +added to the JSON-RPC member projection. The existing `ipAddress` is the most +recent authenticated reached host; the manual-node worker follows it for +cluster-derived entries, so existing node detail shows the reached address. +User-saved entries retain their address, name, kind and precedence. + +At most **8** distinct endpoints are retained per member, most recent successful +first, with the pairing-time endpoint retained as a fallback. Learned endpoints +expire after **30 days** without authenticated success (untried claims age from +first receipt; repeating a claim does not renew it). Only numeric unicast IPs +and ports 1–65535 are accepted; unspecified, multicast, broadcast, IPv4 link-local +and self-reported loopback addresses are rejected. IPv6 link-local hints use the +receiving/dialing connection's **local** zone, never the sender's interface name. + +The existing reachability chooser orders attempts within the whole-request +budget; TCP acceptance alone never proves identity. At most **one unproven +candidate per member per 30 seconds** is admitted to a dial walk, across +concurrent reconciles and restarts, rotating fairly across untried hints. +Redirects are not followed. Every application connection still verifies the +member's exact current pin. A hostile address can fail, not impersonate a peer. +Removal/rejoin is serialized with learning and clears all hints; a request or +response already in flight cannot restore removed state. + +At least one authenticated rendezvous path must remain: for example a peer +reaches the other's saved overlay address and teaches its own current addresses. +If both machines lose every known path, re-establish reachability explicitly; +this is not an unauthenticated discovery or naming service. Endorsements still +fan out certificate trust, but a third party's relayed address grants no +address-learning authority. + ## Cluster trust fan-out Pairing is point-to-point, but trust is **not** left point-to-point: every diff --git a/services/nvpair-cluster-manager/addresses.go b/services/nvpair-cluster-manager/addresses.go new file mode 100644 index 000000000..56a6f00e6 --- /dev/null +++ b/services/nvpair-cluster-manager/addresses.go @@ -0,0 +1,283 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "log" + "net" + "net/netip" + "slices" + "sort" + "strconv" + "strings" + "time" + + "nvpair-shared/netpick" +) + +const ( + maxPeerAddresses = 8 // includes the permanent pairing-time fallback + peerAddressTTL = 30 * 24 * time.Hour + newAddressInterval = 30 * time.Second +) + +// Address hints are local durable state, not membership authority and not part +// of ClusterNode's JSON-RPC projection. Only cluster-manager writes them. +type addressCandidate struct { + Address string `json:"address"` + LearnedAt int64 `json:"learnedAt"` + LastSuccess int64 `json:"lastSuccess,omitempty"` + LastAttempt int64 `json:"lastAttempt,omitempty"` +} + +type memberAddresses struct { + PairingAddress string `json:"pairingAddress,omitempty"` + AddressCandidates []addressCandidate `json:"addressCandidates,omitempty"` + NextAddressProbe int64 `json:"nextAddressProbe,omitempty"` +} + +type persistedMember struct { + ClusterNode + memberAddresses +} + +// candidateEndpoint accepts only numeric unicast addresses, never a hostname +// whose subsequent resolution could redirect a learned hint. A zone is local +// routing state: accept it only on IPv6 link-local addresses. Self-reported +// zones are replaced with the receiving connection's zone before validation. +func candidateEndpoint(endpoint string, allowLoopback bool) (string, bool) { + host, p, err := net.SplitHostPort(endpoint) + if err != nil { + return "", false + } + port, err := strconv.Atoi(p) + if err != nil || port < 1 || port > 65535 { + return "", false + } + ip, err := netip.ParseAddr(host) + if err != nil { + return "", false + } + ip = ip.Unmap() + if ip.IsUnspecified() || ip.IsMulticast() || (!ip.IsGlobalUnicast() && !ip.IsLinkLocalUnicast() && !ip.IsLoopback()) || (!allowLoopback && ip.IsLoopback()) { + return "", false + } + if ip.Is4() { + b := ip.As4() + if b[0] == 0 || b[0] >= 224 || ip.IsLinkLocalUnicast() { + return "", false + } + } + if ip.Is6() && ip.IsLinkLocalUnicast() && ip.Zone() == "" { + return "", false + } + if zone := ip.Zone(); zone != "" { + if !ip.Is6() || !ip.IsLinkLocalUnicast() || len(zone) > 64 || strings.ContainsAny(zone, "%/\\?#[]: \t\r\n") { + return "", false + } + } + return net.JoinHostPort(ip.String(), strconv.Itoa(port)), true +} + +func normalizeMemberAddresses(n *ClusterNode, now time.Time) { + a := &n.addresses + if a.PairingAddress == "" && len(a.AddressCandidates) == 0 { + a.PairingAddress = joinHostPort(n.IPAddress, n.Port, n.Port) + } + if addr, ok := candidateEndpoint(a.PairingAddress, true); ok { + a.PairingAddress = addr + } else { + a.PairingAddress = "" + } + seen := make(map[string]bool) + kept := make([]addressCandidate, 0, maxPeerAddresses-1) + for _, c := range a.AddressCandidates { + addr, ok := candidateEndpoint(c.Address, true) + if !ok || seen[addr] || max(c.LearnedAt, c.LastSuccess) < now.Add(-peerAddressTTL).UnixMilli() { + continue + } + seen[addr] = true + c.Address = addr + // Disk state is also bounded and sanitized, including unreasonable clocks. + c.LearnedAt = min(c.LearnedAt, now.UnixMilli()) + c.LastSuccess = min(c.LastSuccess, now.UnixMilli()) + kept = append(kept, c) + } + sort.SliceStable(kept, func(i, j int) bool { return kept[i].LastSuccess > kept[j].LastSuccess }) + a.AddressCandidates = kept[:min(len(kept), maxPeerAddresses-1)] + // A stale display/dial address must not bypass candidate expiry or the cap. + current := joinHostPort(n.IPAddress, n.Port, n.Port) + if current != a.PairingAddress && !slices.ContainsFunc(a.AddressCandidates, func(c addressCandidate) bool { return c.Address == current && c.LastSuccess > 0 }) { + addr := a.PairingAddress + if len(a.AddressCandidates) > 0 && a.AddressCandidates[0].LastSuccess > 0 { + addr = a.AddressCandidates[0].Address + } + if addr != "" { + n.IPAddress, n.Port = splitAddr(addr, n.Port) + } else { + n.IPAddress = "" + } + } + a.NextAddressProbe = min(a.NextAddressProbe, now.Add(newAddressInterval).UnixMilli()) +} + +func memberCandidateAddrs(n ClusterNode, now time.Time) []string { + normalizeMemberAddresses(&n, now) + out := make([]string, 0, maxPeerAddresses) + for _, c := range n.addresses.AddressCandidates { + out = append(out, c.Address) + } + if n.addresses.PairingAddress != "" && !slices.Contains(out, n.addresses.PairingAddress) { + out = append(out, n.addresses.PairingAddress) + } + return out +} + +// learnPeerAddressesLocked is called only after rechecking the exact connection +// pin and admission under rosterMu. An authenticated sender can teach only its +// own member record. No removed member is created, and no mDNS/preview/gossip +// address reaches this function. observed carries the receiver's IPv6 zone. +func (m *Manager) learnPeerAddressesLocked(uuid, observed string, reports []string, successAddr string, now time.Time) bool { + pin, pinned := m.trust.Get(uuid) + if !pinned || m.teardownPending.Load() { + return false + } + m.memMu.Lock() + defer m.memMu.Unlock() + n, ok := m.members[uuid] + if !ok || n.State != stateMember || n.AdmissionEpoch != pin.AdmissionEpoch || n.ClusterID != pin.ClusterID { + return false + } + normalizeMemberAddresses(n, now) + before := *cloneClusterNode(n) + add := func(addr string, success bool) { + if addr == n.addresses.PairingAddress && !success { + return + } + i := slices.IndexFunc(n.addresses.AddressCandidates, func(c addressCandidate) bool { return c.Address == addr }) + if i >= 0 { + if !success { + return + } // a repeated claim never renews expiry or retry budget + c := n.addresses.AddressCandidates[i] + n.addresses.AddressCandidates = append(n.addresses.AddressCandidates[:i], n.addresses.AddressCandidates[i+1:]...) + c.LastSuccess = now.UnixMilli() + n.addresses.AddressCandidates = append([]addressCandidate{c}, n.addresses.AddressCandidates...) + } else { + c := addressCandidate{Address: addr, LearnedAt: now.UnixMilli()} + if success { + c.LastSuccess = now.UnixMilli() + n.addresses.AddressCandidates = append([]addressCandidate{c}, n.addresses.AddressCandidates...) + } else if len(n.addresses.AddressCandidates) < maxPeerAddresses-1 { + n.addresses.AddressCandidates = append(n.addresses.AddressCandidates, c) + } + } + n.addresses.AddressCandidates = n.addresses.AddressCandidates[:min(len(n.addresses.AddressCandidates), maxPeerAddresses-1)] + if success { + n.IPAddress, n.Port = splitAddr(addr, n.Port) + } + } + if observed != "" { + if addr, ok := candidateEndpoint(joinHostPort(observed, n.Port, m.port), true); ok { + add(addr, true) + } + } + if addr, ok := candidateEndpoint(successAddr, true); ok { + add(addr, true) + } + // Bound processing as well as storage; a large body cannot churn the set. + via := observed + if via == "" { + via = hostOnly(successAddr) + } + for _, addr := range reports[:min(len(reports), maxPeerAddresses)] { + addr = localScope(addr, via) + if addr, ok := candidateEndpoint(addr, false); ok { + add(addr, false) + } + } + return before.IPAddress != n.IPAddress || before.Port != n.Port || !slices.Equal(before.addresses.AddressCandidates, n.addresses.AddressCandidates) +} + +// limitCandidateDials reserves at most one unproven endpoint per member per +// 30 seconds, shared by concurrent reconciles and persisted across restarts. +// A peer rotating its claims cannot reset this per-member budget. Previously +// authenticated sources/successes and the pairing fallback remain available. +func (m *Manager) limitCandidateDials(uuid string, addrs []string, now time.Time) ([]string, bool) { + m.memMu.Lock() + defer m.memMu.Unlock() + n, ok := m.members[uuid] + if !ok { + return nil, false + } + normalizeMemberAddresses(n, now) + allowed := make(map[string]bool) + for _, c := range n.addresses.AddressCandidates { + if c.LastSuccess > 0 { + allowed[c.Address] = true + } + } + allowed[n.addresses.PairingAddress] = true + var probe string + if now.UnixMilli() >= n.addresses.NextAddressProbe { + // Oldest attempted hint first prevents a dead first claim starving the rest. + oldest := int64(1<<63 - 1) + for _, c := range n.addresses.AddressCandidates { + if c.LastSuccess == 0 && slices.Contains(addrs, c.Address) && c.LastAttempt < oldest { + probe = c.Address + oldest = c.LastAttempt + } + } + if probe != "" { + n.addresses.NextAddressProbe = now.Add(newAddressInterval).UnixMilli() + for i := range n.addresses.AddressCandidates { + if n.addresses.AddressCandidates[i].Address == probe { + n.addresses.AddressCandidates[i].LastAttempt = now.UnixMilli() + } + } + allowed[probe] = true + } + } + out := make([]string, 0, maxPeerAddresses) + for _, addr := range addrs { + if allowed[addr] && !slices.Contains(out, addr) { + out = append(out, addr) + } + if len(out) == maxPeerAddresses { + break + } + } + return out, probe != "" +} + +// A reservation must reach disk before a new endpoint is dialed. Otherwise +// repeated restarts during a write failure could bypass the per-member budget. +func (m *Manager) reservePeerDialCandidates(uuid string, addrs []string) []string { + m.rosterMu.Lock() + defer m.rosterMu.Unlock() + allowed, reserved := m.limitCandidateDials(uuid, addrs, time.Now()) + if reserved { + if err := m.persistMembersErr(); err != nil { + log.Printf("persist address dial budget: %v", err) + // The in-memory reservation is retained. A second filter excludes + // unproven hints while leaving proven endpoints/fallback available. + allowed, _ = m.limitCandidateDials(uuid, allowed, time.Now()) + } + } + return allowed +} + +func localRosterAddresses(port int) []string { + var out []string + // The shared publisher includes routed overlay addresses behind the LAN. + for _, host := range netpick.LocalCandidates(netpick.Evidence{}) { + if addr, ok := candidateEndpoint(net.JoinHostPort(host, strconv.Itoa(port)), false); ok { + out = append(out, addr) + } + if len(out) == maxPeerAddresses { + break + } + } + return out +} diff --git a/services/nvpair-cluster-manager/addresses_test.go b/services/nvpair-cluster-manager/addresses_test.go new file mode 100644 index 000000000..4a9aad179 --- /dev/null +++ b/services/nvpair-cluster-manager/addresses_test.go @@ -0,0 +1,151 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "crypto/tls" + "crypto/x509" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "slices" + "strings" + "testing" + "time" +) + +func TestRoamingCandidateValidationAndIPv6Zones(t *testing.T) { + for _, addr := range []string{"0.0.0.0:1", "[::]:1", "255.255.255.255:1", "224.0.0.1:1", "[ff02::1%en0]:1", "[fe80::1]:1", "[2001:db8::1%en0]:1", "[fe80::1%bad/name]:1", "127.0.0.1:1", "169.254.1.2:1", "example.test:1", "10.0.0.1:65536", "10.0.0.1:0"} { + if got, ok := candidateEndpoint(addr, false); ok { + t.Errorf("accepted %q as %q", addr, got) + } + } + for _, addr := range []string{"100.64.0.1:14321", "192.168.1.1:14321", "[2001:db8::1]:14321", "[fe80::1%en0]:14321"} { + if got, ok := candidateEndpoint(addr, false); !ok || got != addr { + t.Errorf("rejected/changed %q: %q %v", addr, got, ok) + } + } + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingRequest(t, m, b, "[fe80::2%local0]:50000", []string{"[fe80::3%remote0]:14999"}) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + got := m.resolvePeerAddrs(*n) + if !slices.Contains(got, "[fe80::3%local0]:14999") || slices.Contains(got, "[fe80::3%remote0]:14999") { + t.Fatalf("remote scope leaked into dial: %v", got) + } + if !strings.Contains(peerURL("https", "[fe80::3%local0]:14999", rosterPath), "%25local0") { + t.Fatal("IPv6 zone not URL escaped") + } +} + +func TestRoamingNewCandidateDialBudgetSurvivesClaimChurnAndReload(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + hints := []string{"100.64.0.1:14999", "100.64.0.2:14999", "100.64.0.3:14999"} + roamingRequest(t, m, b, "10.0.0.1:50000", hints) + now := time.Now() + n, _ := m.memberByNodeID(b.identity.NodeUUID) + first, reserved := m.limitCandidateDials(n.NodeUUID, m.resolvePeerAddrs(*n), now) + if !reserved || len(first) != 3 || !slices.Contains(first, hints[0]) { + t.Fatalf("first probe: %v %v", first, reserved) + } + m.persistMembers() + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + roamingRequest(t, m, b, "10.0.0.1:50000", []string{"100.64.0.4:14999"}) + n, _ = m.memberByNodeID(b.identity.NodeUUID) + second, reserved := m.limitCandidateDials(n.NodeUUID, m.resolvePeerAddrs(*n), now.Add(time.Second)) + if reserved || len(second) != 2 { + t.Fatalf("new claims/restart reset rate limit: %v %v", second, reserved) + } + third, reserved := m.limitCandidateDials(n.NodeUUID, m.resolvePeerAddrs(*n), now.Add(31*time.Second)) + if !reserved || !slices.Contains(third, hints[1]) { + t.Fatalf("untried hint starved: %v %v", third, reserved) + } +} + +type roamingBlockedBody struct { + entered, release chan struct{} + reader io.Reader +} + +func (b *roamingBlockedBody) Read(p []byte) (int, error) { + select { + case <-b.entered: + default: + close(b.entered) + } + <-b.release + return b.reader.Read(p) +} +func (b *roamingBlockedBody) Close() error { return nil } + +func TestRoamingInFlightRequestCannotRestoreRemovedMember(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + blocked := &roamingBlockedBody{entered: make(chan struct{}), release: make(chan struct{}), reader: strings.NewReader(`{"clusterId":"cluster-1","selfAddresses":["100.64.0.8:14999"]}`)} + r := httptest.NewRequest(http.MethodPost, rosterPath, nil) + r.Body = blocked + r.RemoteAddr = "10.0.0.8:50000" + cert, err := x509.ParseCertificate(b.identity.Cert.Certificate[0]) + if err != nil { + t.Fatal(err) + } + r.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{cert}} + w := httptest.NewRecorder() + done := make(chan struct{}) + go func() { defer close(done); m.handleRoster(w, r) }() + <-blocked.entered + m.rosterMu.Lock() + if err := m.trust.Remove(b.identity.NodeUUID); err != nil { + t.Fatal(err) + } + m.removeMemberByUUID(b.identity.NodeUUID) + m.persistMembers() + m.rosterMu.Unlock() + close(blocked.release) + <-done + if w.Code != http.StatusForbidden { + t.Fatalf("stale request accepted: %d", w.Code) + } + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + if _, exists := m.memberByNodeID(b.identity.NodeUUID); exists { + t.Fatal("in-flight learning resurrected removed member") + } +} + +func TestRoamingLearnsSelfReportFromPinnedResponseOnly(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + const hint = "100.64.0.8:14999" + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if _, ok := b.verifyClientPin(r); !ok { + http.Error(w, "forbidden", 403) + return + } + _ = json.NewEncoder(w).Encode(map[string]interface{}{"clusterId": "cluster-1", "selfAddresses": []string{hint}}) + })) + pinTrusted(t, b, m.identity.NodeUUID, string(m.identity.CertPEM), m.identity.CertFingerprint) + srv.TLS = b.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: b.identity.NodeUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + outcome, _ := m.reconcileWithin(context.Background(), []string{addr}, b.identity.NodeUUID) + if outcome != reconcileAccepted { + t.Fatalf("outcome=%v", outcome) + } + n, _ := m.memberByNodeID(b.identity.NodeUUID) + if !slices.Contains(m.resolvePeerAddrs(*n), hint) { + t.Fatal("pinned response self-report not learned") + } +} diff --git a/services/nvpair-cluster-manager/connreuse_test.go b/services/nvpair-cluster-manager/connreuse_test.go index 83254c579..354b7b9f7 100644 --- a/services/nvpair-cluster-manager/connreuse_test.go +++ b/services/nvpair-cluster-manager/connreuse_test.go @@ -64,6 +64,8 @@ func TestReconcile_ReusesPeerConnections(t *testing.T) { defer ts.Close() addr := ts.Listener.Addr().String() + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: peerUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) const rounds = 5 for i := 0; i < rounds; i++ { if outcome, _ := m.reconcileWith([]string{addr}, peerUUID); outcome != reconcileAccepted { diff --git a/services/nvpair-cluster-manager/httpserver.go b/services/nvpair-cluster-manager/httpserver.go index 7eb71ec5d..f59eb1e0c 100644 --- a/services/nvpair-cluster-manager/httpserver.go +++ b/services/nvpair-cluster-manager/httpserver.go @@ -507,16 +507,8 @@ func (m *Manager) prepareJoinerInitialComplete(inviteID string, sess *pairingSes log.Printf("pairing %s: cannot read inviter PairingInfo: %v", inviteID, err) return nil } - // The inviter's advertised link-local address carries the zone of its own - // interface; give it ours before it is dialled or recorded as the member's - // address, and keep it as an observation so a roster claim of the inviter's - // spelling cannot replace it. - if scoped := localScope(pi.Addr, inviterIP); scoped != pi.Addr { - pi.Addr = scoped - if host, _, err := net.SplitHostPort(scoped); err == nil { - m.noteObservedPeerHost(pi.NodeUUID, host) - } - } + // Remote zones name the peer's interfaces; scope to the incoming link. + pi.Addr = localScope(pi.Addr, inviterIP) sess.peerPairing = pi sess.peerCert = cert sess.addr = pi.Addr @@ -656,7 +648,6 @@ func (m *Manager) onInviterPaired(inviteID string, sess *pairingSession) error { host, port := splitAddr(pi.Addr, m.port) if sess.peerSrcIP != "" { host = sess.peerSrcIP - m.noteObservedPeerHost(pi.NodeUUID, host) } if host != "" { pi.Addr = net.JoinHostPort(host, strconv.Itoa(port)) diff --git a/services/nvpair-cluster-manager/manager.go b/services/nvpair-cluster-manager/manager.go index bef58bd04..9dc3819ab 100644 --- a/services/nvpair-cluster-manager/manager.go +++ b/services/nvpair-cluster-manager/manager.go @@ -82,9 +82,10 @@ type Manager struct { // wait is never erased by a now-stale unanimous-403 verdict. clusterGen atomic.Uint64 - memMu sync.Mutex - members map[string]*ClusterNode // keyed by nodeUuid - invites map[string]*Invite // keyed by inviteId + membersWriteMu sync.Mutex // serialize durable snapshots, including address learning + memMu sync.Mutex + members map[string]*ClusterNode // keyed by nodeUuid + invites map[string]*Invite // keyed by inviteId tsMu sync.Mutex tombstones map[string]Tombstone // keyed by removed nodeUuid; newest wins (fan-out removal) @@ -132,13 +133,6 @@ type Manager struct { browser *Browser - // observedPeerHosts maps a member's uuid to the source host it was last seen - // connecting from over an authenticated transport. Memory-only: it is - // re-earned by the first inbound reconcile after a restart, and a stale - // persisted copy would be a claim rather than an observation. - observedMu sync.Mutex - observedPeerHosts map[string]string - // peerAddrs remembers which of a member's advertised addresses answers, so a // multi-homed peer is reconciled with at an address that works rather than at // whichever one it happens to rank first. diff --git a/services/nvpair-cluster-manager/membership.go b/services/nvpair-cluster-manager/membership.go index d207ab792..00ac48b5a 100644 --- a/services/nvpair-cluster-manager/membership.go +++ b/services/nvpair-cluster-manager/membership.go @@ -61,6 +61,8 @@ type ClusterNode struct { State MembershipState `json:"state"` JoinedAt *int64 `json:"joinedAt"` LastSeen *int64 `json:"lastSeen"` + + addresses memberAddresses // local-only durable hints; never serialized to IPC } // Invite is the Broker-facing view of a pairing session (§6). Certificates are @@ -88,6 +90,7 @@ func cloneClusterNode(node *ClusterNode) *ClusterNode { return nil } cp := *node + cp.addresses.AddressCandidates = append([]addressCandidate(nil), node.addresses.AddressCandidates...) if node.JoinedAt != nil { v := *node.JoinedAt cp.JoinedAt = &v @@ -122,7 +125,11 @@ func cloneInvite(inv *Invite) *Invite { // upsertMember inserts or replaces a member, keyed by nodeUuid. func (m *Manager) upsertMember(node *ClusterNode) { m.memMu.Lock() - m.members[node.NodeUUID] = cloneClusterNode(node) + copy := cloneClusterNode(node) + if copy.State == stateMember { + normalizeMemberAddresses(copy, time.Now()) + } + m.members[node.NodeUUID] = copy m.memMu.Unlock() m.clusterGen.Add(1) } @@ -133,6 +140,7 @@ func (m *Manager) removeMemberByUUID(uuid string) bool { _, ok := m.members[uuid] delete(m.members, uuid) m.memMu.Unlock() + m.peerAddrs.Forget(uuid) if ok { m.clusterGen.Add(1) } @@ -142,8 +150,12 @@ func (m *Manager) removeMemberByUUID(uuid string) bool { // resetMembership clears the entire member set (used when leaving a cluster). func (m *Manager) resetMembership() { m.memMu.Lock() + old := m.members m.members = make(map[string]*ClusterNode) m.memMu.Unlock() + for uuid := range old { + m.peerAddrs.Forget(uuid) + } m.clusterGen.Add(1) } @@ -179,68 +191,6 @@ func (m *Manager) memberByNodeID(idOrUUID string) (*ClusterNode, bool) { return nil, false } -// setMemberAddr updates a member's reachable host:port under the membership -// lock, returning whether anything actually changed. A zero/invalid port keeps -// the member's existing listening port (falling back to the local default), so -// callers learning only a host (e.g. a TCP source IP, whose port is ephemeral) -// don't clobber the real listening port. Callers persist and emit -// nodes:changed when this returns true. -func (m *Manager) setMemberAddr(uuid, host string, port int) bool { - if host == "" { - return false - } - m.memMu.Lock() - defer m.memMu.Unlock() - n, ok := m.members[uuid] - if !ok { - return false - } - if port < 1 || port > 65535 { - port = n.Port - } - if port < 1 || port > 65535 { - port = m.port - } - if n.IPAddress == host && n.Port == port { - return false - } - n.IPAddress = host - n.Port = port - return true -} - -// noteObservedPeerHost records the source host an authenticated peer connected -// from and applies it as that member's address, returning whether anything -// changed. -// -// An address observed on a connection is the only claim about a peer's -// reachability this node has verified: the peer reached us from it, over a -// transport pinned to its certificate. A peer's own advertised address — or worse, -// a third peer's relayed copy of it — is that peer guessing which of its addresses -// works from here, which is exactly the guess that goes wrong on a multi-homed -// host. So the observation is remembered, and refreshMemberAddr will not overwrite -// it with a claim. -func (m *Manager) noteObservedPeerHost(uuid, host string) bool { - if uuid == "" || host == "" { - return false - } - m.observedMu.Lock() - if m.observedPeerHosts == nil { - m.observedPeerHosts = make(map[string]string) - } - m.observedPeerHosts[uuid] = host - m.observedMu.Unlock() - return m.setMemberAddr(uuid, host, 0) -} - -// observedPeerHost returns the source host this peer was last seen connecting -// from, or "" if it never has been. -func (m *Manager) observedPeerHost(uuid string) string { - m.observedMu.Lock() - defer m.observedMu.Unlock() - return m.observedPeerHosts[uuid] -} - // updateMemberIdentity refreshes a member's display fields (id/name) in place, // keyed by its stable nodeUuid. Empty inputs are ignored so a partial update // can't blank a known value. Returns whether anything changed. Display-only: the diff --git a/services/nvpair-cluster-manager/membership_store.go b/services/nvpair-cluster-manager/membership_store.go index f5ac30ab3..6c44f565e 100644 --- a/services/nvpair-cluster-manager/membership_store.go +++ b/services/nvpair-cluster-manager/membership_store.go @@ -11,6 +11,7 @@ import ( "os" "path/filepath" "sort" + "time" ) // membersFile is the durable, lean roster persisted under the cluster dir. Only @@ -30,11 +31,15 @@ func (m *Manager) persistMembers() { } func (m *Manager) persistMembersErr() error { + m.membersWriteMu.Lock() + defer m.membersWriteMu.Unlock() m.memMu.Lock() - durable := make([]ClusterNode, 0, len(m.members)) + durable := make([]persistedMember, 0, len(m.members)) for _, n := range m.members { if n.State == stateMember { - durable = append(durable, *cloneClusterNode(n)) + normalizeMemberAddresses(n, time.Now()) + copy := cloneClusterNode(n) + durable = append(durable, persistedMember{ClusterNode: *copy, memberAddresses: copy.addresses}) } } m.memMu.Unlock() @@ -57,14 +62,16 @@ func (m *Manager) loadMembers() error { if err != nil { return err } - var nodes []ClusterNode + var nodes []persistedMember if err := json.Unmarshal(data, &nodes); err != nil { return err } m.memMu.Lock() defer m.memMu.Unlock() for i := range nodes { - n := nodes[i] + n := nodes[i].ClusterNode + n.addresses = nodes[i].memberAddresses + normalizeMemberAddresses(&n, time.Now()) m.members[n.NodeUUID] = &n } return nil diff --git a/services/nvpair-cluster-manager/mtls.go b/services/nvpair-cluster-manager/mtls.go index 43d6cca3c..e834bd864 100644 --- a/services/nvpair-cluster-manager/mtls.go +++ b/services/nvpair-cluster-manager/mtls.go @@ -50,7 +50,11 @@ func (m *Manager) peerClient(peerUUID string) (*http.Client, error) { if !ok { return nil, fmt.Errorf("no pin for peer %s", peerUUID) } - return client, nil + // Peer responses cannot send candidate dials to a different endpoint or + // downgrade transport. Keep the pooled transport, but prohibit redirects. + bounded := *client + bounded.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + return &bounded, nil } func (m *Manager) buildPairingClientTLSConfig(peerDER []byte) (*tls.Config, error) { @@ -152,20 +156,35 @@ func (m *Manager) handleMembersRemove(w http.ResponseWriter, r *http.Request) { // notifyPeerRemoval best-effort tells a peer it has been removed, over mTLS, // while the peer's pin still exists (call before deleting it locally). -func (m *Manager) notifyPeerRemoval(addr, peerUUID string, proof RemovalProof) { +func (m *Manager) notifyPeerRemoval(node ClusterNode, proof RemovalProof) { + peerUUID := node.NodeUUID client, err := m.peerClient(peerUUID) if err != nil { log.Printf("notify removal: %v", err) return } + ctx, cancel := context.WithTimeout(context.Background(), pairingHTTPTimeout) + defer cancel() + addrs := m.reservePeerDialCandidates(peerUUID, m.resolvePeerAddrs(node)) body, _ := json.Marshal(membersRemoveRequest{NodeUUID: m.identity.NodeUUID, Proof: proof}) - resp, err := client.Post(peerURL("https", addr, membersRemovePath), "application/json", bytes.NewReader(body)) - if err != nil { - log.Printf("notify removal to %s (%s): %v", addr, peerUUID, err) + for _, addr := range confirmedFirst(addrs, m.peerAddrs.ChooseWithin(ctx, peerUUID, addrs)) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, peerURL("https", addr, membersRemovePath), bytes.NewReader(body)) + if err != nil { + continue + } + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + m.peerAddrs.Forget(peerUUID) + if ctx.Err() != nil { + break + } + continue + } + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBodyBytes)) + _ = resp.Body.Close() return } - _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBodyBytes)) - _ = resp.Body.Close() } // serveSplit serves both transport modes on one listener via the shared diff --git a/services/nvpair-cluster-manager/peer_addr_test.go b/services/nvpair-cluster-manager/peer_addr_test.go index 024aff153..12166ba40 100644 --- a/services/nvpair-cluster-manager/peer_addr_test.go +++ b/services/nvpair-cluster-manager/peer_addr_test.go @@ -14,8 +14,6 @@ import ( "sync/atomic" "testing" "time" - - "nvpair-shared/noderec" ) // unreachableAddr is in TEST-NET-1 (RFC 5737), which is reserved for @@ -27,22 +25,22 @@ const unreachableAddr = "192.0.2.1:14321" // could not reach, so every reconcile burned the full request timeout against it // while the address that worked sat unused in the same candidate list. func TestReconcileWithFailsOverToAnAnsweringAddress(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - mA := newTestManagerPort(t, 15031) - mB := newTestManagerPort(t, 15032) - go func() { _ = mA.runHTTP(ctx) }() - go func() { _ = mB.runHTTP(ctx) }() - time.Sleep(400 * time.Millisecond) - + mA, mB := newTestManager(t), newTestManager(t) + srv := httptest.NewUnstartedServer(http.HandlerFunc(mA.handleRoster)) + _, port := splitAddr(srv.Listener.Addr().String(), mA.port) + mA.port = port + srv.TLS = mA.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() pinTrusted(t, mA, mB.identity.NodeUUID, string(mB.identity.CertPEM), mB.identity.CertFingerprint) pinTrusted(t, mB, mA.identity.NodeUUID, string(mA.identity.CertPEM), mA.identity.CertFingerprint) - mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "127.0.0.1", Port: 15032, AdmissionEpoch: 1, State: stateMember}) - mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "192.0.2.1", Port: 15031, AdmissionEpoch: 1, State: stateMember}) - - working := net.JoinHostPort("127.0.0.1", strconv.Itoa(15031)) - outcome, _ := mB.reconcileWith([]string{unreachableAddr, working}, mA.identity.NodeUUID) + mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "127.0.0.1", Port: mB.port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "192.0.2.1", Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + working := strings.TrimPrefix(srv.URL, "https://") + mB.memMu.Lock() + mB.members[mA.identity.NodeUUID].addresses.AddressCandidates = []addressCandidate{{Address: working, LearnedAt: time.Now().UnixMilli()}} + mB.memMu.Unlock() + outcome, _ := mB.reconcileWith([]string{net.JoinHostPort("192.0.2.1", strconv.Itoa(port)), working}, mA.identity.NodeUUID) if outcome != reconcileAccepted { t.Fatalf("reconcile outcome = %v, want accepted via the second candidate", outcome) } @@ -58,6 +56,12 @@ func TestReconcileBudgetCoversTheWholeCandidateWalk(t *testing.T) { pinTrusted(t, m, peerUUID, certPEM, fingerprint) blackholed := []string{"192.0.2.1:14321", "192.0.2.2:14321", "192.0.2.3:14321", "192.0.2.4:14321"} + m.upsertMember(&ClusterNode{NodeUUID: peerUUID, ID: "peer", IPAddress: "192.0.2.1", Port: 14321, State: stateMember}) + m.memMu.Lock() + for _, addr := range blackholed[1:] { + m.members[peerUUID].addresses.AddressCandidates = append(m.members[peerUUID].addresses.AddressCandidates, addressCandidate{Address: addr, LearnedAt: time.Now().UnixMilli()}) + } + m.memMu.Unlock() budget := 600 * time.Millisecond ctx, cancel := context.WithTimeout(context.Background(), budget) defer cancel() @@ -76,102 +80,7 @@ func TestReconcileBudgetCoversTheWholeCandidateWalk(t *testing.T) { } } -// TestResolvePeerAddrsPutsTheRecordedAddressFirst: the pairing-time address was -// observed working from here, so it leads — but the peer's other advertised -// addresses follow it rather than being discarded. -func TestResolvePeerAddrsPutsTheRecordedAddressFirst(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "33333333-3333-3333-3333-333333333333" - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.240.1", - IPs: []string{"192.168.240.1", "10.172.55.129"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - got := m.resolvePeerAddrs(ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "10.172.55.129", Port: 14321}) - want := []string{"10.172.55.129:14321", "192.168.240.1:14321"} - if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { - t.Fatalf("resolvePeerAddrs = %v, want %v", got, want) - } -} - -// TestRefreshMemberAddrsFromMDNSKeepsAStillAdvertisedAddress: the stored address -// was proven from this host, while the peer's own first choice is ranked from its -// vantage point, so a peer that still advertises what we hold must not be -// redirected onto a link only its neighbour can use. -func TestRefreshMemberAddrsFromMDNSKeepsAStillAdvertisedAddress(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "44444444-4444-4444-4444-444444444444" - m.upsertMember(&ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "10.172.55.129", Port: 14321, State: stateMember}) - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.240.1", - IPs: []string{"192.168.240.1", "10.172.55.129"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "10.172.55.129" { - t.Fatalf("stored address = %q, want it kept at 10.172.55.129", n.IPAddress) - } - - // A peer that has stopped advertising the stored address really moved. - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "10.172.55.200", - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "10.172.55.200" { - t.Fatalf("stored address = %q, want the moved peer's 10.172.55.200", n.IPAddress) - } -} - -func TestResolvePeerAddrs_PrefersLocalLANOverRecordedCGNAT(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "55555555-5555-5555-5555-555555555555" - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.1.50", - IPs: []string{"192.168.1.50", "100.64.1.2"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - // Node was recorded with Tailscale CGNAT address. - got := m.resolvePeerAddrs(ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "100.64.1.2", Port: 14321}) - want := []string{"192.168.1.50:14321", "100.64.1.2:14321"} - if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { - t.Fatalf("resolvePeerAddrs = %v, want LAN first %v", got, want) - } -} - -func TestRefreshMemberAddrsFromMDNS_UpgradesCGNATToLAN(t *testing.T) { - m := newTestManagerPort(t, 14321) - peer := "66666666-6666-6666-6666-666666666666" - // Node currently holds Tailscale CGNAT address. - m.upsertMember(&ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "100.64.1.2", Port: 14321, State: stateMember}) - m.browser = newBrowser() - m.browser.setRelay([]noderec.DirectoryNode{{ - HostUUID: peer, - Name: "peer", - IP: "192.168.1.50", - IPs: []string{"192.168.1.50", "100.64.1.2"}, - Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}}, - }}) - - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "192.168.1.50" { - t.Fatalf("stored address = %q, want upgraded to LAN 192.168.1.50", n.IPAddress) - } -} +// Reachability confirmation only reorders hints; authentication remains mTLS. func TestConfirmedFirst(t *testing.T) { addrs := []string{"a:1", "b:1", "c:1"} diff --git a/services/nvpair-cluster-manager/reconnect_test.go b/services/nvpair-cluster-manager/reconnect_test.go index c60760c78..3e649ab11 100644 --- a/services/nvpair-cluster-manager/reconnect_test.go +++ b/services/nvpair-cluster-manager/reconnect_test.go @@ -11,42 +11,9 @@ import ( "time" ) -// TestSetMemberAddr covers the address-update helper: a host-only update keeps -// the stored listening port, identical updates are no-ops, an explicit port is -// honored, and unknown/empty inputs change nothing. -func TestSetMemberAddr(t *testing.T) { - m := newTestManager(t) - uuid := "11111111-1111-1111-1111-111111111111" - m.upsertMember(&ClusterNode{NodeUUID: uuid, ID: "n", IPAddress: "10.0.0.1", Port: 14321, State: stateMember}) +// The real mTLS endpoint learns only the authenticated source host and keeps +// the paired listening port, never the ephemeral client port. - if !m.setMemberAddr(uuid, "192.168.1.50", 0) { - t.Fatal("expected a change for a new host") - } - if n, _ := m.memberByNodeID(uuid); n.IPAddress != "192.168.1.50" || n.Port != 14321 { - t.Fatalf("host-only update wrong: got %s:%d want 192.168.1.50:14321", n.IPAddress, n.Port) - } - if m.setMemberAddr(uuid, "192.168.1.50", 0) { - t.Fatal("identical addr should not report a change") - } - if !m.setMemberAddr(uuid, "192.168.1.50", 14999) { - t.Fatal("expected a change for a new port") - } - if n, _ := m.memberByNodeID(uuid); n.Port != 14999 { - t.Fatalf("explicit port not applied: got %d want 14999", n.Port) - } - if m.setMemberAddr("does-not-exist", "1.2.3.4", 0) { - t.Fatal("unknown uuid should not report a change") - } - if m.setMemberAddr(uuid, "", 0) { - t.Fatal("empty host should not report a change") - } -} - -// TestHandleRosterLearnsSourceAddr drives the real mTLS roster endpoint over -// loopback: the server holds a STALE address for the client (as if the client -// moved since pairing), and after one reconcile it must have learned the -// client's real source IP from the authenticated connection while preserving -// the client's known listening port. func TestHandleRosterLearnsSourceAddr(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -61,8 +28,8 @@ func TestHandleRosterLearnsSourceAddr(t *testing.T) { pinTrusted(t, mB, mA.identity.NodeUUID, string(mA.identity.CertPEM), mA.identity.CertFingerprint) // mA's record for mB points at a stale IP; mB still reaches mA fine. - mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "10.9.9.9", Port: 15022, State: stateMember}) - mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "127.0.0.1", Port: 15021, State: stateMember}) + mA.upsertMember(&ClusterNode{NodeUUID: mB.identity.NodeUUID, ID: "node-b", IPAddress: "10.9.9.9", Port: 15022, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + mB.upsertMember(&ClusterNode{NodeUUID: mA.identity.NodeUUID, ID: "node-a", IPAddress: "127.0.0.1", Port: 15021, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) mB.reconcileWith([]string{net.JoinHostPort("127.0.0.1", strconv.Itoa(15021))}, mA.identity.NodeUUID) @@ -81,29 +48,3 @@ func TestHandleRosterLearnsSourceAddr(t *testing.T) { // TestRefreshMemberAddrsFromMDNS verifies the discovery backstop: a member // whose stored address is stale is refreshed from the mDNS browse map, the // pass is a safe no-op without a browser, and the self entry is never rewritten. -func TestRefreshMemberAddrsFromMDNS(t *testing.T) { - m := newTestManager(t) - peer := "22222222-2222-2222-2222-222222222222" - m.upsertMember(&ClusterNode{NodeUUID: peer, ID: "peer", IPAddress: "10.9.9.9", Port: 14321, State: stateMember}) - - // No browser configured yet: must not panic and must change nothing. - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "10.9.9.9" { - t.Fatalf("addr changed without a browser: got %q", n.IPAddress) - } - - m.browser = newBrowser() - m.browser.seed(peer, "192.168.1.77", 14321) - m.refreshMemberAddrsFromMDNS() - if n, _ := m.memberByNodeID(peer); n.IPAddress != "192.168.1.77" { - t.Fatalf("mDNS refresh did not update stored addr: got %q want 192.168.1.77", n.IPAddress) - } - - // The self entry must never be redirected by (untrusted) mDNS. - m.addSelfMember() - m.browser.seed(m.identity.NodeUUID, "5.5.5.5", 14321) - m.refreshMemberAddrsFromMDNS() - if self, _ := m.memberByNodeID(m.identity.NodeUUID); self.IPAddress != "127.0.0.1" { - t.Fatalf("self address must not be rewritten from mDNS: got %q", self.IPAddress) - } -} diff --git a/services/nvpair-cluster-manager/remove.go b/services/nvpair-cluster-manager/remove.go index 80b604660..1d703dbaa 100644 --- a/services/nvpair-cluster-manager/remove.go +++ b/services/nvpair-cluster-manager/remove.go @@ -6,8 +6,6 @@ package main import ( "encoding/json" "fmt" - "net" - "strconv" ) type removeParams struct { @@ -92,9 +90,7 @@ func (m *Manager) handleNodesRemove(msg *Message) { } // Notify first, while the peer's pin is still present to authenticate mTLS. - if node.IPAddress != "" { - m.notifyPeerRemoval(net.JoinHostPort(node.IPAddress, strconv.Itoa(node.Port)), node.NodeUUID, proof) - } + m.notifyPeerRemoval(*node, proof) // De-pin and drop membership under the cluster-composition boundary so the // removal is atomic against a concurrent self-remove teardown — the two can't // interleave and leave a half-cleared roster. The network notify above stays diff --git a/services/nvpair-cluster-manager/roaming_order_test.go b/services/nvpair-cluster-manager/roaming_order_test.go new file mode 100644 index 000000000..1f9a71406 --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_order_test.go @@ -0,0 +1,20 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "slices" + "testing" +) + +func TestRoamingSuccessfulPairingFallbackBecomesMostRecent(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingRequest(t, m, b, "10.0.0.2:50000", nil) + roamingRequest(t, m, b, "192.0.2.20:50000", nil) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + if got := m.resolvePeerAddrs(*n); !slices.Equal(got, []string{"192.0.2.20:14999", "10.0.0.2:14999"}) { + t.Fatalf("most recent success not first: %v", got) + } +} diff --git a/services/nvpair-cluster-manager/roaming_persistence_test.go b/services/nvpair-cluster-manager/roaming_persistence_test.go new file mode 100644 index 000000000..60559b188 --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_persistence_test.go @@ -0,0 +1,35 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "net/http" + "net/http/httptest" + "os" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestRoamingNewCandidateNeedsDurableDialBudget(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + var hits atomic.Int32 + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hits.Add(1); w.WriteHeader(200) })) + srv.TLS = b.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + m.memMu.Lock() + m.members[b.identity.NodeUUID].addresses.AddressCandidates = []addressCandidate{{Address: addr, LearnedAt: time.Now().UnixMilli()}} + m.memMu.Unlock() + if err := os.Mkdir(m.membersPath(), 0700); err != nil { + t.Fatal(err) + } // atomic rename must fail + m.reconcileWith([]string{addr}, b.identity.NodeUUID) + if hits.Load() != 0 { + t.Fatal("dialed unproven hint without persisting its retry budget") + } +} diff --git a/services/nvpair-cluster-manager/roaming_redirect_test.go b/services/nvpair-cluster-manager/roaming_redirect_test.go new file mode 100644 index 000000000..dde48d374 --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_redirect_test.go @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" +) + +func TestRoamingPinnedPeerCannotRedirectCandidateDials(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + var hits atomic.Int32 + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hits.Add(1) })) + defer target.Close() + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL, http.StatusTemporaryRedirect) + })) + srv.TLS = b.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: b.identity.NodeUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + m.reconcileWith([]string{addr}, b.identity.NodeUUID) + if hits.Load() != 0 { + t.Fatal("pinned peer redirected a dial outside the candidate budget") + } +} diff --git a/services/nvpair-cluster-manager/roaming_test.go b/services/nvpair-cluster-manager/roaming_test.go new file mode 100644 index 000000000..4713aed1b --- /dev/null +++ b/services/nvpair-cluster-manager/roaming_test.go @@ -0,0 +1,188 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "bytes" + "crypto/tls" + "crypto/x509" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "slices" + "strconv" + "strings" + "testing" + "time" +) + +func roamingMember(t *testing.T, m, peer *Manager) { + t.Helper() + pinTrusted(t, m, peer.identity.NodeUUID, string(peer.identity.CertPEM), peer.identity.CertFingerprint) + m.upsertMember(&ClusterNode{ID: peer.identity.NodeID, NodeUUID: peer.identity.NodeUUID, Name: "peer", IPAddress: "192.0.2.20", Port: peer.port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) +} + +func roamingRequest(t *testing.T, m, peer *Manager, source string, hints []string) int { + t.Helper() + body, err := json.Marshal(map[string]interface{}{"clusterId": "cluster-1", "selfAddresses": hints}) + if err != nil { + t.Fatal(err) + } + r := httptest.NewRequest(http.MethodPost, rosterPath, bytes.NewReader(body)) + cert, err := x509.ParseCertificate(peer.identity.Cert.Certificate[0]) + if err != nil { + t.Fatal(err) + } + r.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{cert}} + r.RemoteAddr = source + w := httptest.NewRecorder() + m.handleRoster(w, r) + return w.Code +} + +func TestRoamingSelfReportsBelongOnlyToPinnedSender(t *testing.T) { + m, b, c, unknown := newTestManager(t), newTestManager(t), newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingMember(t, m, c) + hints := []string{"100.64.0.8:14999", "[2001:db8::8]:14999"} + if code := roamingRequest(t, m, unknown, "10.0.0.8:50000", hints); code != http.StatusForbidden { + t.Fatalf("unknown cert: %d", code) + } + if code := roamingRequest(t, m, c, "10.0.0.9:50000", hints); code != http.StatusOK { + t.Fatalf("pinned cert: %d", code) + } + bn, _ := m.memberByNodeID(b.identity.NodeUUID) + if got := m.resolvePeerAddrs(*bn); !slices.Equal(got, []string{"192.0.2.20:14999"}) { + t.Fatalf("another cert taught B addresses: %v", got) + } + cn, _ := m.memberByNodeID(c.identity.NodeUUID) + got := m.resolvePeerAddrs(*cn) + if !slices.Contains(got, hints[0]) || !slices.Contains(got, hints[1]) { + t.Fatalf("verified self-report was not learned: %v", got) + } +} + +func TestRoamingCandidatesBoundedOrderedPersistedAndExpired(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + var hints []string + for i := 1; i <= 30; i++ { + hints = append(hints, "100.64.1."+strconv.Itoa(i)+":14999") + } + // Many valid endpoints, followed by the most recently observed connection. + hints = append(hints, "100.64.0.1:14999", "100.64.0.2:14999", "100.64.0.3:14999", "100.64.0.4:14999", "100.64.0.5:14999", "100.64.0.6:14999", "100.64.0.7:14999", "100.64.0.8:14999", "100.64.0.9:14999") + roamingRequest(t, m, b, "10.0.0.1:50000", hints) + roamingRequest(t, m, b, "10.0.0.2:50000", nil) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + got := m.resolvePeerAddrs(*n) + if len(got) != 8 || got[0] != "10.0.0.2:14999" || got[1] != "10.0.0.1:14999" || got[len(got)-1] != "192.0.2.20:14999" { + t.Fatalf("candidate order/bound/fallback: %v", got) + } + if err := m.persistMembersErr(); err != nil { + t.Fatal(err) + } + st, err := os.Stat(m.membersPath()) + if err != nil { + t.Fatal(err) + } + if st.Mode().Perm() != 0600 { + t.Fatalf("permissions: %v", st.Mode()) + } + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + n, _ = m.memberByNodeID(b.identity.NodeUUID) + if after := m.resolvePeerAddrs(*n); !slices.Equal(after, got) { + t.Fatalf("reload: %v != %v", after, got) + } + // Age every learned hint through the actual persistence format, then reload. + data, _ := os.ReadFile(m.membersPath()) + var stored []struct { + ClusterNode + PairingAddress string `json:"pairingAddress"` + AddressCandidates []struct { + Address string `json:"address"` + LearnedAt int64 `json:"learnedAt"` + LastSuccess int64 `json:"lastSuccess"` + } `json:"addressCandidates"` + } + if err := json.Unmarshal(data, &stored); err != nil { + t.Fatal(err) + } + if len(stored) != 1 || len(stored[0].AddressCandidates) == 0 { + t.Fatal("learned candidates not persisted") + } + for i := range stored[0].AddressCandidates { + stored[0].AddressCandidates[i].LearnedAt = time.Now().Add(-31 * 24 * time.Hour).UnixMilli() + stored[0].AddressCandidates[i].LastSuccess = stored[0].AddressCandidates[i].LearnedAt + } + data, _ = json.Marshal(stored) + if err := os.WriteFile(m.membersPath(), data, 0600); err != nil { + t.Fatal(err) + } + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + n, _ = m.memberByNodeID(b.identity.NodeUUID) + if after := m.resolvePeerAddrs(*n); !slices.Equal(after, []string{"192.0.2.20:14999"}) { + t.Fatalf("expired hints survived: %v", after) + } + m.removeMemberByUUID(b.identity.NodeUUID) + m.persistMembers() + roamingRequest(t, m, b, "10.0.0.3:50000", hints) + if _, exists := m.memberByNodeID(b.identity.NodeUUID); exists { + t.Fatal("removed member resurrected by hint") + } +} + +func TestRoamingIgnoresDiscoveryAndRelayedAddresses(t *testing.T) { + m, b, c := newTestManager(t), newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingMember(t, m, c) + m.browser = newBrowser() + m.browser.seed(b.identity.NodeUUID, "10.0.0.99", 14321) + m.mergeRoster(&Roster{ClusterID: "cluster-1", Members: []RosterEntry{{NodeUUID: b.identity.NodeUUID, Addr: "10.0.0.98:14321"}}}, c.identity.NodeUUID) + n, _ := m.memberByNodeID(b.identity.NodeUUID) + if got := m.resolvePeerAddrs(*n); !slices.Equal(got, []string{"192.0.2.20:14999"}) { + t.Fatalf("unauthenticated/relayed address became learned authority: %v", got) + } +} + +func TestRoamingHostileHintCannotRedirectPinnedIdentity(t *testing.T) { + m, b, hostile := newTestManager(t), newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + hits := 0 + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + _ = json.NewEncoder(w).Encode(hostile.buildLocalRoster()) + })) + srv.TLS = hostile.buildServerTLSConfig() + srv.StartTLS() + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "https://") + host, port := splitAddr(addr, m.port) + m.upsertMember(&ClusterNode{NodeUUID: b.identity.NodeUUID, ID: "peer", IPAddress: host, Port: port, ClusterID: "cluster-1", AdmissionEpoch: 1, State: stateMember}) + outcome, _ := m.reconcileWith([]string{addr}, b.identity.NodeUUID) + if outcome != reconcileUnreachable || hits != 0 { + t.Fatalf("hostile address redirected identity: outcome=%v hits=%d", outcome, hits) + } +} + +func TestRoamingInboundSourceSurvivesRestart(t *testing.T) { + m, b := newTestManager(t), newTestManager(t) + roamingMember(t, m, b) + roamingRequest(t, m, b, "10.0.0.2:50000", nil) + m.resetMembership() + if err := m.loadMembers(); err != nil { + t.Fatal(err) + } + n, _ := m.memberByNodeID(b.identity.NodeUUID) + got := m.resolvePeerAddrs(*n) + if !slices.Equal(got, []string{"10.0.0.2:14999", "192.0.2.20:14999"}) { + t.Fatalf("restart lost source or pairing fallback: %v", got) + } +} diff --git a/services/nvpair-cluster-manager/roster.go b/services/nvpair-cluster-manager/roster.go index d33207a6e..511b33bdd 100644 --- a/services/nvpair-cluster-manager/roster.go +++ b/services/nvpair-cluster-manager/roster.go @@ -36,6 +36,7 @@ type Roster struct { Members []RosterEntry `json:"members"` Tombstones []Tombstone `json:"tombstones,omitempty"` // legacy compatibility RemovalProofs []RemovalProof `json:"removalProofs,omitempty"` + SelfAddresses []string `json:"selfAddresses,omitempty"` // only the authenticated sender, never relayed } // selfPub returns this node's own Ed25519 public key. @@ -59,6 +60,7 @@ func (m *Manager) buildLocalRoster() *Roster { cid, _ := m.clusterIdentity() r := &Roster{ ClusterID: cid, + SelfAddresses: localRosterAddresses(m.port), Tombstones: m.snapshotTombstones(), RemovalProofs: m.snapshotRemovalProofs(), } @@ -107,6 +109,10 @@ func (m *Manager) buildLocalRoster() *Roster { func (m *Manager) mergeRoster(remote *Roster, senderUUID string) bool { m.rosterMu.Lock() defer m.rosterMu.Unlock() + return m.mergeRosterLocked(remote, senderUUID) +} + +func (m *Manager) mergeRosterLocked(remote *Roster, senderUUID string) bool { if m.teardownPending.Load() { return false } @@ -221,7 +227,7 @@ func (m *Manager) applyMembers(entries []RosterEntry, cid, senderUUID string) bo if !m.entryEndorsed(e, cid) { continue } - if err := m.pinEntry(e, der); err != nil { + if err := m.pinEntry(e, der, senderUUID); err != nil { continue } if err := m.clearRemovalProofAfterAdmission(e.NodeUUID, e.AdmissionEpoch); err != nil { @@ -235,7 +241,6 @@ func (m *Manager) applyMembers(entries []RosterEntry, cid, senderUUID string) bo if err := m.trust.AddEndorsements(e.NodeUUID, e.Endorsements); err != nil { log.Printf("roster: persist endorsements for %s: %v", e.NodeUUID, err) } - m.refreshMemberAddr(e) if e.NodeUUID == senderUUID && m.refreshMemberIdentity(e) { changed = true } @@ -251,7 +256,7 @@ func (m *Manager) applyMembers(entries []RosterEntry, cid, senderUUID string) bo if !m.entryEndorsed(e, cid) { continue // no endorsement from a node we currently trust } - if err := m.pinEntry(e, der); err != nil { + if err := m.pinEntry(e, der, senderUUID); err != nil { continue } if err := m.clearRemovalProofAfterAdmission(e.NodeUUID, e.AdmissionEpoch); err != nil { @@ -307,9 +312,15 @@ func (m *Manager) supersededByTombstone(e *RosterEntry) bool { } // pinEntry pins an accepted roster entry's cert and records it as a member. -func (m *Manager) pinEntry(e *RosterEntry, der []byte) error { +func (m *Manager) pinEntry(e *RosterEntry, der []byte, senderUUID string) error { now := time.Now().UnixMilli() - host, port := splitAddr(e.Addr, m.port) + // Endorsements authenticate membership, not a relayed address claim. + host, port := "", m.port + if e.NodeUUID == senderUUID { + if addr, ok := candidateEndpoint(e.Addr, false); ok { + host, port = splitAddr(addr, m.port) + } + } cid, _ := m.clusterIdentity() previous, hadPrevious := m.memberByNodeID(e.NodeUUID) m.upsertMember(&ClusterNode{ @@ -350,25 +361,6 @@ func (m *Manager) pinEntry(e *RosterEntry, der []byte) error { return nil } -// refreshMemberAddr updates a known member's reachable address from a roster -// entry without disturbing its other fields. -// -// A host this node has seen the member connect from outranks the entry's claim, -// which may be the member's own guess about which of its addresses works from -// here — or another peer's relayed copy of that guess. The port is taken from the -// entry either way: a listening port is a fact the member knows and an observer -// cannot infer, since the source port of an inbound connection is ephemeral. -func (m *Manager) refreshMemberAddr(e *RosterEntry) { - if e.Addr == "" { - return - } - host, port := splitAddr(e.Addr, m.port) - if observed := m.observedPeerHost(e.NodeUUID); observed != "" { - host = observed - } - m.setMemberAddr(e.NodeUUID, host, port) -} - // refreshMemberIdentity updates a still-trusted peer's display id/name — in both // the member record and the persisted pin — from its own roster entry, keyed by // the stable nodeUuid. This tracks a peer's PC rename so the roster and UI show diff --git a/services/nvpair-cluster-manager/roster_http.go b/services/nvpair-cluster-manager/roster_http.go index f78425b91..c15db309c 100644 --- a/services/nvpair-cluster-manager/roster_http.go +++ b/services/nvpair-cluster-manager/roster_http.go @@ -10,11 +10,8 @@ import ( "io" "log" "net/http" - "slices" "sync" "time" - - "nvpair-shared/netpick" ) // reconcileEvery is the heartbeat cadence for catch-up roster reconciles — the @@ -43,14 +40,21 @@ func (m *Manager) handleRoster(w http.ResponseWriter, r *http.Request) { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } + // Snapshot the pin and both admissions before reading an untrusted body. + // Removal/rejoin may run while that read blocks; recheck before any merge + // or learning, under the same lock as membership removal. + m.rosterMu.Lock() senderUUID, ok := m.verifyClientPin(r) - if !ok { + pin, pinned := m.trust.Get(senderUUID) + cid0, epoch0 := m.currentAdmission() + m.rosterMu.Unlock() + if !ok || !pinned { m.rejectRosterReconcile(w, r) return } - body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes)) - if err != nil { - http.Error(w, "read body", http.StatusBadRequest) + body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes+1)) + if err != nil || len(body) > maxBodyBytes { + http.Error(w, "invalid body", http.StatusBadRequest) return } var remote Roster @@ -58,20 +62,25 @@ func (m *Manager) handleRoster(w http.ResponseWriter, r *http.Request) { http.Error(w, "invalid JSON", http.StatusBadRequest) return } - changed := m.mergeRoster(&remote, senderUUID) + m.rosterMu.Lock() + cid, epoch := m.currentAdmission() + current, stillPinned := m.trust.Get(senderUUID) + _, valid := m.verifyClientPin(r) + if !valid || !stillPinned || current.AdmissionEpoch != pin.AdmissionEpoch || cid == "" || cid != cid0 || epoch != epoch0 { + m.rosterMu.Unlock() + http.Error(w, "stale peer", http.StatusForbidden) + return + } + changed := m.mergeRosterLocked(&remote, senderUUID) + if remote.ClusterID == cid { + changed = m.learnPeerAddressesLocked(senderUUID, hostOnly(r.RemoteAddr), remote.SelfAddresses, "", time.Now()) || changed + } + // Persist under the composition boundary: an old in-flight request must + // never write learned addresses back after member removal. if changed { - log.Printf("roster: reconciled from peer %s; membership changed", senderUUID) - } - // Learn the peer's current address from the authenticated connection so a - // member that moved to a new IP becomes reachable again without waiting on - // mDNS. verifyClientPin already authenticated the sender against its pinned - // cert, so the TCP source IP is trustworthy enough to dial back on the - // member's known listening port (we pass port 0 to keep the stored - // listening port, never the ephemeral source port). - if host := hostOnly(r.RemoteAddr); host != "" && m.noteObservedPeerHost(senderUUID, host) { - log.Printf("roster: refreshed peer %s address to %s (from connection)", senderUUID, host) - changed = true + m.persistMembers() } + m.rosterMu.Unlock() if changed { m.emitNodesChanged() } @@ -214,6 +223,7 @@ func (m *Manager) reconcileWith(addrs []string, peerUUID string) (reconcileOutco // reconcileWithin is reconcileWith under a caller-supplied budget. func (m *Manager) reconcileWithin(ctx context.Context, addrs []string, peerUUID string) (reconcileOutcome, bool) { + addrs = m.reservePeerDialCandidates(peerUUID, addrs) if len(addrs) == 0 { return reconcileUnreachable, false } @@ -265,6 +275,13 @@ func confirmedFirst(addrs []string, confirmed string) []string { // reconcileWith to try the next candidate; every actual response — including a // rejection — is that peer's answer and ends the walk. func (m *Manager) reconcileOnce(ctx context.Context, client *http.Client, body []byte, addr, peerUUID string) (outcome reconcileOutcome, provenRemoval, answered bool) { + m.rosterMu.Lock() + cid0, epoch0 := m.currentAdmission() + pin0, pinned := m.trust.Get(peerUUID) + m.rosterMu.Unlock() + if !pinned { + return reconcileUnreachable, false, false + } req, err := http.NewRequestWithContext(ctx, http.MethodPost, peerURL("https", addr, rosterPath), bytes.NewReader(body)) if err != nil { log.Printf("roster: build reconcile request for %s (%s): %v", addr, peerUUID, err) @@ -277,7 +294,17 @@ func (m *Manager) reconcileOnce(ctx context.Context, client *http.Client, body [ return reconcileUnreachable, false, false } defer resp.Body.Close() - rb, _ := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes)) + rb, readErr := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes+1)) + m.rosterMu.Lock() + defer m.rosterMu.Unlock() + cid, epoch := m.currentAdmission() + pin, stillPinned := m.trust.Get(peerUUID) + if !stillPinned || pin.AdmissionEpoch != pin0.AdmissionEpoch || cid == "" || cid != cid0 || epoch != epoch0 || m.teardownPending.Load() || resp.TLS == nil || len(resp.TLS.PeerCertificates) == 0 || !m.trust.MatchDER(peerUUID, resp.TLS.PeerCertificates[0].Raw) { + return reconcileUnreachable, false, false + } + if readErr != nil || len(rb) > maxBodyBytes { + return reconcileUnreachable, false, true + } // A 403 is an authenticated signal (we pinned the peer's server cert, so we // know it is really that peer) that it no longer holds our pin. But a dropped // pin alone does not prove *we* were removed — a peer that itself left drops @@ -302,8 +329,11 @@ func (m *Manager) reconcileOnce(ctx context.Context, client *http.Client, body [ log.Printf("roster: decode reconcile response from %s: %v", peerUUID, err) return reconcileAccepted, false, true // 200: the peer accepted us; only the response body was unusable } - if m.mergeRoster(&remote, peerUUID) { - log.Printf("roster: reconciled with peer %s; membership changed", peerUUID) + changed := m.mergeRosterLocked(&remote, peerUUID) + if remote.ClusterID == cid { + changed = m.learnPeerAddressesLocked(peerUUID, "", remote.SelfAddresses, addr, time.Now()) || changed + } + if changed { m.emitNodesChanged() } return reconcileAccepted, false, true @@ -467,12 +497,8 @@ func (m *Manager) removeBareRejectingPeers(clusterID string, generation uint64, } } -// reconcileLoop runs the periodic catch-up reconcile and tombstone GC until the -// context is cancelled. Each pass first refreshes member addresses from the -// mDNS browse map, so a member that moved to a new IP (sleep/wake, DHCP) -// re-converges within a heartbeat without a dedicated discovery callback. An -// initial pass shortly after startup converges a node that just (re)started or -// woke at a new address without waiting a full heartbeat. +// reconcileLoop runs catch-up reconciliation and tombstone GC until cancelled. +// Only pinned roster connections can update a member's durable address hints. func (m *Manager) reconcileLoop(ctx context.Context) { ticker := time.NewTicker(reconcileEvery) defer ticker.Stop() @@ -490,8 +516,8 @@ func (m *Manager) reconcileLoop(ctx context.Context) { } } -// reconcilePass GCs expired tombstones, refreshes member addresses from the -// discovery resolver, and pushes a roster reconcile to every confirmed peer. +// reconcilePass GCs expired tombstones and pushes a roster reconcile to every +// confirmed peer, using its bounded authenticated address hints. func (m *Manager) reconcilePass() { m.gcTombstones() if cid, _ := m.clusterIdentity(); cid == "" { @@ -499,98 +525,11 @@ func (m *Manager) reconcilePass() { } m.mesh.Refresh() m.clients.DropUnpinned() - m.refreshMemberAddrsFromMDNS() m.reconcilePeersAndMaybeSelfRemove() } -// refreshMemberAddrsFromMDNS updates each confirmed peer's stored reachable -// address from the mDNS browse map when the peer has stopped advertising the -// address we hold. Inter-node calls are mTLS cert-pinned, so trusting mDNS only -// to *locate* a peer is safe — a wrong address simply fails the pinned handshake -// and the next pass corrects it. This is the backstop that re-converges a moved -// member even when it can't reach us to announce its new address itself. -// -// A peer that still advertises the address we hold keeps it. That address was -// observed working from here, whereas the peer's own first choice is ranked from -// its vantage point and may be a link only its cabled neighbour can use, so -// overwriting on every pass would trade a known-good endpoint for a guess. -func (m *Manager) refreshMemberAddrsFromMDNS() { - if m.browser == nil { - return - } - changed := false - for _, n := range m.snapshotNodes() { - if n.NodeUUID == m.identity.NodeUUID || n.State != stateMember { - continue - } - hosts, port, ok := m.browser.Resolve(n.NodeUUID) - if !ok { - hosts, port, ok = m.browser.Resolve(n.ID) - } - if !ok || len(hosts) == 0 { - continue - } - host := hosts[0] - if slices.Contains(hosts, n.IPAddress) { - // A peer that still advertises what we hold keeps it, unless hosts[0] is - // a strictly higher-ranked address class (e.g. physical private LAN over - // Tailscale / CGNAT). This allows seamless upgrade to local Wi-Fi when - // reconnecting to the local network. - if netpick.ScoreAddress(hosts[0]) <= netpick.ScoreAddress(n.IPAddress) { - host = n.IPAddress - } - } - if m.setMemberAddr(n.NodeUUID, host, port) { - log.Printf("roster: refreshed peer %s address to %s (from discovery)", n.NodeUUID, joinHostPort(host, port, m.port)) - changed = true - } - } - if changed { - m.emitNodesChanged() - } -} - -// resolvePeerAddrs returns a member's host:port candidates in the order to try -// them: the address recorded at pairing time first (authoritative — it is the -// source IP the peer connected from plus the listening port it advertised), then -// everything the peer currently advertises through mDNS. -// -// The recorded address leads on equal footing, but if it is a lower-ranked class -// (e.g. CGNAT/Tailscale) than an available physical LAN address from discovery, -// discovery's top candidate leads so local Wi-Fi is preferred when reachable. +// resolvePeerAddrs uses only local, authenticated address hints. Discovery and +// relayed names/addresses cannot modify this set. func (m *Manager) resolvePeerAddrs(n ClusterNode) []string { - addrs := make([]string, 0, 4) - seen := make(map[string]bool, 4) - add := func(addr string) { - if addr == "" || seen[addr] { - return - } - seen[addr] = true - addrs = append(addrs, addr) - } - var hosts []string - var port int - if m.browser != nil { - var ok bool - hosts, port, ok = m.browser.Resolve(n.NodeUUID) - if !ok { - hosts, port, ok = m.browser.Resolve(n.ID) - } - } - leadWithRecorded := n.IPAddress != "" - if leadWithRecorded && len(hosts) > 0 { - if netpick.ScoreAddress(hosts[0]) > netpick.ScoreAddress(n.IPAddress) { - leadWithRecorded = false - } - } - if leadWithRecorded { - add(joinHostPort(n.IPAddress, n.Port, m.port)) - } - for _, h := range hosts { - add(joinHostPort(h, port, m.port)) - } - if !leadWithRecorded && n.IPAddress != "" { - add(joinHostPort(n.IPAddress, n.Port, m.port)) - } - return addrs + return memberCandidateAddrs(n, time.Now()) } diff --git a/services/nvpair-manual-nodes/README.md b/services/nvpair-manual-nodes/README.md index 4358b7eec..d110cf94b 100644 --- a/services/nvpair-manual-nodes/README.md +++ b/services/nvpair-manual-nodes/README.md @@ -11,6 +11,11 @@ Node-info JSON decoding reads at most 1 MiB, and model-list decoding at most 16 MiB. A JSON value that extends past its limit follows the existing probe or model-parse failure path. +Cluster-derived entries follow cluster-manager's authenticated reached address +when a member roams. User-saved PAIR/engine entries retain their original +address, name and intent; the learned roster never overwrites them. Each new +endpoint still needs a pinned-mTLS inventory proof before it becomes routable. + ## Worker control health An id-bearing `worker/health` request (no params) returns `{"ok":true}` with diff --git a/services/nvpair-manual-nodes/manager.go b/services/nvpair-manual-nodes/manager.go index e7f3c3bef..5880523a4 100644 --- a/services/nvpair-manual-nodes/manager.go +++ b/services/nvpair-manual-nodes/manager.go @@ -577,8 +577,10 @@ func (m *Manager) syncClusterMembers() { } found := false for _, member := range activeMembers { - if (tn.status.HostUUID != "" && tn.status.HostUUID == member.NodeUUID) || - tn.entry.Address == member.IPAddress { + // A cluster seed follows the member's reached endpoint. A saved + // entry is never rebuilt here, so its operator-selected address wins. + if tn.entry.Address == member.IPAddress && + (tn.status.HostUUID == "" || tn.status.HostUUID == member.NodeUUID) { found = true break } diff --git a/services/nvpair-manual-nodes/roaming_test.go b/services/nvpair-manual-nodes/roaming_test.go new file mode 100644 index 000000000..f7c7aa2ec --- /dev/null +++ b/services/nvpair-manual-nodes/roaming_test.go @@ -0,0 +1,63 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" +) + +func TestClusterSeedFollowsRoamingAddressWithoutOverwritingSavedEntry(t *testing.T) { + for _, saved := range []bool{false, true} { + name := "cluster seed" + if saved { + name = "saved alias" + } + t.Run(name, func(t *testing.T) { + m, _, _ := newTestManager(t) + m.clusterDir = t.TempDir() + const uuid = "22222222-2222-2222-2222-222222222222" + write := func(addr string, tick int) { + t.Helper() + data, err := json.Marshal([]clusterMemberEntry{{ID: "peer", Name: "peer", NodeUUID: uuid, IPAddress: addr, State: "member"}}) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(m.clusterDir, "members.json") + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + at := time.Now().Add(time.Duration(tick) * time.Second) + if err := os.Chtimes(path, at, at); err != nil { + t.Fatal(err) + } + m.syncClusterMembers() + } + // Start with a previously proven tracked peer, without a background + // probe racing the address-change interleaving being tested. + entry := ManualEntry{Kind: "pair", Address: "192.0.2.20", Name: "peer"} + if saved { + entry.Name = "my-saved-peer" + } + m.mu.Lock() + m.nodes[nodeID(entry)] = &trackedNode{entry: entry, fromCluster: !saved, status: ManualNodeStatus{ID: nodeID(entry), Kind: "pair", Address: entry.Address, HostUUID: uuid, ClusterUUID: uuid, FromCluster: !saved}} + m.mu.Unlock() + write("100.64.0.8", 2) + nodes := m.listNodes() + if len(nodes) != 1 { + t.Fatalf("nodes=%+v", nodes) + } + want := "100.64.0.8" + if saved { + want = "192.0.2.20" + } + if nodes[0].Address != want { + t.Fatalf("reached address=%q want %q", nodes[0].Address, want) + } + }) + } +} diff --git a/services/tests/peer_roaming_test.go b/services/tests/peer_roaming_test.go new file mode 100644 index 000000000..3a9a2b457 --- /dev/null +++ b/services/tests/peer_roaming_test.go @@ -0,0 +1,105 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package tests + +import ( + "encoding/json" + "os" + "path/filepath" + "strconv" + "testing" + "time" +) + +// Real isolated workers: corrupt A's old address for B, let B's pinned inbound +// roster connection repair it, restart A, then prove A reaches B by delivering +// authenticated removal. No multicast prerequisite or environment skip. +func TestPeerRoamingLearnsInboundAndDialsAfterRestart(t *testing.T) { + dirA, dirB := t.TempDir(), t.TempDir() + portA, portB := freePort(t), freePort(t) + a, b := startCM(t, dirA, portA), startCM(t, dirB, portB) + defer func() { + if a != nil { + a.stop() + } + if b != nil { + b.stop() + } + }() + waitForPort(t, "127.0.0.1", portA, 5*time.Second) + waitForPort(t, "127.0.0.1", portB, 5*time.Second) + pairNodes(t, a, b, portB, "peer-b") + bid := decodeResult[cmNodeID](t, b.call("cluster:get-node-id", nil)).NodeUUID + a.stop() + a = nil + b.stop() + b = nil + type diskMember struct { + NodeUUID string `json:"nodeUuid"` + IPAddress string `json:"ipAddress"` + PairingAddress string `json:"pairingAddress"` + AddressCandidates []struct { + Address string `json:"address"` + } `json:"addressCandidates"` + } + path := filepath.Join(dirA, "cluster", "members.json") + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var raw []map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + t.Fatal(err) + } + stale := "192.0.2.20:" + strconv.Itoa(portB) + for _, member := range raw { + var uuid string + _ = json.Unmarshal(member["nodeUuid"], &uuid) + if uuid == bid { + member["ipAddress"], _ = json.Marshal("192.0.2.20") + member["pairingAddress"], _ = json.Marshal(stale) + delete(member, "addressCandidates") + } + } + data, _ = json.Marshal(raw) + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + a = startCM(t, dirA, portA) + b = startCM(t, dirB, portB) + reached := false + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + data, err = os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var members []diskMember + if err := json.Unmarshal(data, &members); err != nil { + t.Fatal(err) + } + for _, member := range members { + if member.NodeUUID != bid || member.IPAddress != "127.0.0.1" { + continue + } + if member.PairingAddress != stale || len(member.AddressCandidates) == 0 { + t.Fatalf("repair lost pairing fallback or durable learned source: %+v", member) + } + reached = true + } + if reached { + break + } + time.Sleep(100 * time.Millisecond) + } + a.stop() + a = nil + if !reached { + t.Fatal("B's authenticated inbound connection never repaired A's stale member address") + } + a = startCM(t, dirA, portA) + waitForPort(t, "127.0.0.1", portA, 5*time.Second) + a.call("nodes:remove", map[string]string{"nodeUuid": bid}) + waitMembers(t, b, 0) +} diff --git a/services/versions.json b/services/versions.json index 2f5c0cfe3..80ae55402 100644 --- a/services/versions.json +++ b/services/versions.json @@ -7,13 +7,13 @@ "lmstudio-proxy": "1.5.0", "nvpair-node-info": "1.4.0", "nvpair-node-scanner": "1.3.0", - "nvpair-manual-nodes": "1.3.0", + "nvpair-manual-nodes": "1.3.1", "nvpair-workload-manager": "1.3.1", "nvpair-errors": "1.1.1", "nvpair-node-settings": "3.2.0", "nvpair-ui-broker": "2.5.0", "nvpair-engine-manager": "1.4.0", - "nvpair-cluster-manager": "2.2.0", + "nvpair-cluster-manager": "2.3.0", "nvpair-job-scheduler": "1.1.0", "nvpair-tui": "1.3.0", "omlx-proxy": "1.5.0" From bfb09d9b493e9913d97bd56de9bd9e9dfea9cac6 Mon Sep 17 00:00:00 2001 From: Jared Ping <13906915+JaredP94@users.noreply.github.com> Date: Fri, 6 Nov 2026 12:15:56 +0200 Subject: [PATCH 2/2] fix(scanner): wait for local node-info readiness before warning Scanner starts before node-info has necessarily opened its listener. Give only the first local enrichment read a bounded fifteen-second readiness budget using the existing telemetry backoff. Normal startup stays quiet; exhaustion still warns and subsequent outages retain immediate transition reporting. Cover delayed readiness and a sustained outage, and bump the scanner patch version. Signed-off-by: Jared Ping <13906915+JaredP94@users.noreply.github.com> --- services/nvpair-node-scanner/README.md | 6 ++ services/nvpair-node-scanner/daemon.go | 4 +- .../nvpair-node-scanner/local_readiness.go | 47 +++++++++++ .../local_readiness_test.go | 84 +++++++++++++++++++ services/versions.json | 2 +- 5 files changed, 141 insertions(+), 2 deletions(-) create mode 100644 services/nvpair-node-scanner/local_readiness.go create mode 100644 services/nvpair-node-scanner/local_readiness_test.go diff --git a/services/nvpair-node-scanner/README.md b/services/nvpair-node-scanner/README.md index 3e00e89e8..5bb208868 100644 --- a/services/nvpair-node-scanner/README.md +++ b/services/nvpair-node-scanner/README.md @@ -9,6 +9,12 @@ SPDX-License-Identifier: Apache-2.0 Concretely, the daemon: +- **Waits for local node-info on startup:** only the first local inventory read + retries with the existing 2, 4, 8-second telemetry backoff, under a single + 15-second request-and-wait budget. A normal listener startup logs no outage; + a real outage still warns when that budget ends. After the first successful + read (or exhausted startup budget), failures warn immediately as before. + - **Advertises** this node's one `_nvpair-node._tcp` record, built from a registry of the local services its parent (the broker) registers — `ni`/`ol`/`lm`/`er`/`wl`/`cl`/`em`/`ec` ports plus the node identity (`uuid=`, `cluster-uuid=` when clustered, `ip=`). One record per node covers every service. - **Browses** `_nvpair-node._tcp` for candidate endpoints. Verified/local records are keyed by authenticated host UUID; unverified previews have independent `preview:` record keys and an explicit `identityVerification: "unverified"` marker. A claimed UUID cannot overwrite or remove a verified record. - **Authenticates before enrichment:** remote hardware and broker-owned service metadata come from `ni` over pinned cluster mTLS, with matching certificate/response/metadata identities and an exact-current-pin check after reading. Only authenticated metadata supplies service ports. Models, per-engine attribution and loaded state then come from the authenticated `em` port over pinned mTLS. Local inventory remains an owned loopback read. Public HTTP hardware is display-only and never supplies service models or scheduler telemetry. diff --git a/services/nvpair-node-scanner/daemon.go b/services/nvpair-node-scanner/daemon.go index 789ced50f..71502a6cf 100644 --- a/services/nvpair-node-scanner/daemon.go +++ b/services/nvpair-node-scanner/daemon.go @@ -100,6 +100,8 @@ type daemon struct { // into a torn self entry. The self key is written nowhere else — onBrowse // ignores our own uuid — so this is the sole self-key writer lock. selfMu sync.Mutex + // Only the first local enrichment waits for node-info's startup readiness. + localInfoStartup sync.Once // lastInfo / lastModels cache each node's last successful enrichment (keyed by // hostUuid) — GPU/CPU/memory from node-info and the model list from @@ -1231,7 +1233,7 @@ func (d *daemon) enrichInfoCandidates(node *noderec.DirectoryNode, hosts []strin return } ask := func(host string) (NodeInfoResponse, bool) { - candidate, fetched := d.fetchNodeInfo(host, ni.Port) + candidate, fetched := d.fetchLocalNodeInfo(host, ni.Port, node.HostUUID) if !fetched { return NodeInfoResponse{}, false } diff --git a/services/nvpair-node-scanner/local_readiness.go b/services/nvpair-node-scanner/local_readiness.go new file mode 100644 index 000000000..5f5b89d36 --- /dev/null +++ b/services/nvpair-node-scanner/local_readiness.go @@ -0,0 +1,47 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "time" +) + +const localNodeInfoStartupTimeout = 15 * time.Second + +// fetchLocalNodeInfo gives only the first local inventory read a readiness +// grace. The broker spawns scanner before node-info's listener is ready. Reuse +// the existing telemetry backoff, bounded across requests and waits, so normal +// startup is quiet but a real outage still reaches noteNodeInfo's warning. +// Later outages retain the immediate transition report and last-good cache. +func (d *daemon) fetchLocalNodeInfo(host string, port int, expectedUUID string) (NodeInfoResponse, bool) { + var info NodeInfoResponse + var ok, initial bool + d.localInfoStartup.Do(func() { + initial = true + if d.http == nil && d.tlsHTTP == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), localNodeInfoStartupTimeout) + defer cancel() + for attempt := 0; ctx.Err() == nil; attempt++ { + info, ok = d.fetchNodeInfoWithin(ctx, host, port) + if ok && (info.HostUUID == "" || info.HostUUID == expectedUUID) { + return + } + ok = false + timer := time.NewTimer(telemetryRetryDelay(attempt)) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } + }) + if initial { + return info, ok + } + return d.fetchNodeInfo(host, port) +} diff --git a/services/nvpair-node-scanner/local_readiness_test.go b/services/nvpair-node-scanner/local_readiness_test.go new file mode 100644 index 000000000..1a36558f1 --- /dev/null +++ b/services/nvpair-node-scanner/local_readiness_test.go @@ -0,0 +1,84 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "nvpair-shared/noderec" +) + +func localReadinessFixture(t *testing.T, handler http.Handler) (*daemon, *noderec.DirectoryNode) { + t.Helper() + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + u, err := url.Parse(srv.URL) + if err != nil { + t.Fatal(err) + } + port, err := strconv.Atoi(u.Port()) + if err != nil { + t.Fatal(err) + } + d := newSelfTestDaemon("self", "127.0.0.1") + return d, &noderec.DirectoryNode{HostUUID: "self", Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceNodeInfo: {Port: port}}} +} + +func TestLocalNodeInfoStartupWaitsBeforeWarning(t *testing.T) { + logs := captureLogs(t) + var requests atomic.Int32 + d, node := localReadinessFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if requests.Add(1) == 1 { + http.Error(w, "starting", http.StatusServiceUnavailable) + return + } + _ = json.NewEncoder(w).Encode(NodeInfoResponse{HostUUID: "self"}) + })) + d.enrichInfoAt(node, "127.0.0.1") + if requests.Load() < 2 { + t.Fatal("startup did not retry local node-info readiness") + } + if _, ok := d.lastInfo["self"]; !ok { + t.Fatal("local inventory not read after startup") + } + if strings.Contains(logs.String(), "not answering") { + t.Fatalf("normal startup warned: %s", logs.String()) + } + // Once it answered, a later outage must warn immediately, not get a new grace. + d.http.CloseIdleConnections() + d.http = &http.Client{Transport: readinessUnavailable{}} + start := time.Now() + d.enrichInfoAt(node, "127.0.0.1") + if !strings.Contains(logs.String(), "not answering") || time.Since(start) > time.Second { + t.Fatalf("later outage was hidden/delayed: %s", logs.String()) + } +} + +type readinessUnavailable struct{} + +func (readinessUnavailable) RoundTrip(*http.Request) (*http.Response, error) { + return nil, http.ErrServerClosed +} + +func TestLocalNodeInfoRealStartupOutageEventuallyWarns(t *testing.T) { + logs := captureLogs(t) + d, node := localReadinessFixture(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Error(w, "down", 503) })) + start := time.Now() + d.enrichInfoAt(node, "127.0.0.1") + elapsed := time.Since(start) + if elapsed < 10*time.Second || elapsed > 20*time.Second { + t.Fatalf("startup readiness wait=%v, want bounded grace", elapsed) + } + if got := strings.Count(logs.String(), "not answering"); got != 1 { + t.Fatalf("outage warnings=%d: %s", got, logs.String()) + } +} diff --git a/services/versions.json b/services/versions.json index 80ae55402..0a255a572 100644 --- a/services/versions.json +++ b/services/versions.json @@ -6,7 +6,7 @@ "ollama-proxy": "1.5.0", "lmstudio-proxy": "1.5.0", "nvpair-node-info": "1.4.0", - "nvpair-node-scanner": "1.3.0", + "nvpair-node-scanner": "1.3.1", "nvpair-manual-nodes": "1.3.1", "nvpair-workload-manager": "1.3.1", "nvpair-errors": "1.1.1",