Preserve contradictory saved cluster trust through producer and refresh

Carry exact duplicate-row conflicts into owned client trust before map deduplication. Keep unresolved saved trust through initialization, missing-client retry and topology refresh without replacing pins or stopping trusted siblings. Cover actual monitor lifecycle routes and member projections alongside client TLS controls.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -42,9 +42,25 @@ through normalized authorities, invalid nonempty pins and a capture race.
 `TestClusterConflictingPinsEndpointProjection` uses the production endpoint
 builder, real clients and monitor projection in all three auth modes: native
 Online, identity and saved trust are not changed by failed Pulse reachability.
+The producer also records contradictory exact duplicate saved rows before
+its map/deduplication can overwrite them, and carries those endpoint conflicts
+through initialisation, missing-client reconnect and cluster refresh. The
+cluster owns a copy of that conflict set; aliases cannot escape it, and a
+later repeated row cannot clear it. Normal configuration reload with agreeing
+pins is required to resume that endpoint. A non-conflicting sibling is not
+refused. The public constructor's optional conflict inputs preserve existing
+callers and do not introduce a new trust mode. Producer tests cover both
+routes, pin formatting, repeated contradictions and configuration ownership.
+The projection cases include exact duplicates as well as URL aliases.
+A credential-free agreement check also holds automatic topology refresh while
+saved trust is contradictory: discovery cannot collapse the rows, move an
+alias or substitute a pin to clear that refusal. Current trusted sibling
+clients keep monitoring; topology changes wait for independently reconciled
+configuration. `TestClusterPinConflictsSurviveMonitorLifecycle` exercises the
+actual initialisation, missing-client retry and refresh/rebuild paths in all
+three auth modes and both layouts, including normal refresh after reconciliation.
 This establishes source/consumer behaviour, not native recovery or attribution
-of a guest, PBS or certificate report. Exact duplicate saved rows collapsed by
-the builder's map are outside this URL-alias resolver contract.
+of a guest, PBS or certificate report.
 
 
 ### Cluster recovery requires success of the operation actually probed

docs/release-control/v6/internal/subsystems/security-privacy.md

@@ -43,9 +43,25 @@ through normalized authorities, invalid nonempty pins and a capture race.
 `TestClusterConflictingPinsEndpointProjection` uses the production endpoint
 builder, real clients and monitor projection in all three auth modes: native
 Online, identity and saved trust are not changed by failed Pulse reachability.
+The producer also records contradictory exact duplicate saved rows before
+its map/deduplication can overwrite them, and carries those endpoint conflicts
+through initialisation, missing-client reconnect and cluster refresh. The
+cluster owns a copy of that conflict set; aliases cannot escape it, and a
+later repeated row cannot clear it. Normal configuration reload with agreeing
+pins is required to resume that endpoint. A non-conflicting sibling is not
+refused. The public constructor's optional conflict inputs preserve existing
+callers and do not introduce a new trust mode. Producer tests cover both
+routes, pin formatting, repeated contradictions and configuration ownership.
+The projection cases include exact duplicates as well as URL aliases.
+A credential-free agreement check also holds automatic topology refresh while
+saved trust is contradictory: discovery cannot collapse the rows, move an
+alias or substitute a pin to clear that refusal. Current trusted sibling
+clients keep monitoring; topology changes wait for independently reconciled
+configuration. `TestClusterPinConflictsSurviveMonitorLifecycle` exercises the
+actual initialisation, missing-client retry and refresh/rebuild paths in all
+three auth modes and both layouts, including normal refresh after reconciliation.
 This establishes source/consumer behaviour, not native recovery or attribution
-of a guest, PBS or certificate report. Exact duplicate saved rows collapsed by
-the builder's map are outside this URL-alias resolver contract.
+of a guest, PBS or certificate report.
 
 
 ### First-use cluster trust includes password construction

internal/monitoring/monitor_additional_test.go

@@ -335,7 +335,7 @@ func TestBuildClusterEndpointsForInit_RespectsDiscoveryPolicy(t *testing.T) {
 		},
 	}
 
-	endpoints, _ := monitor.buildClusterEndpointsForInit(config.PVEInstance{
+	endpoints, _, _ := monitor.buildClusterEndpointsForInit(config.PVEInstance{
 		Name:      "cluster-a",
 		Host:      "https://main.local:8006",
 		VerifySSL: true,

internal/monitoring/monitor_client_init.go

@@ -29,7 +29,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) {
 
 		// Check if this is a cluster
 		if pve.IsCluster && len(pve.ClusterEndpoints) > 0 {
-			endpoints, endpointFingerprints := m.buildClusterEndpointsForInit(pve)
+			endpoints, endpointFingerprints, conflictingEndpoints := m.buildClusterEndpointsForInit(pve)
 
 			log.Info().
 				Str("cluster", pve.ClusterName).
@@ -44,6 +44,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) {
 				clientConfig,
 				endpoints,
 				endpointFingerprints,
+				conflictingEndpoints...,
 			)
 			m.replacePVEClient(pve.Name, clusterClient)
 			log.Info().

internal/monitoring/monitor_client_reconnect.go

@@ -108,11 +108,11 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) {
 			}
 			if pve.IsCluster && len(pve.ClusterEndpoints) > 0 {
 				// Create cluster client
-				endpoints, endpointFingerprints := m.buildClusterEndpointsForReconnect(pve)
+				endpoints, endpointFingerprints, conflictingEndpoints := m.buildClusterEndpointsForReconnect(pve)
 
 				clientConfig := config.CreateProxmoxConfig(&pve)
 				clientConfig.Timeout = connectionTimeout
-				clusterClient, err := proxmox.NewClusterClientWithContext(ctx, pve.Name, clientConfig, endpoints, endpointFingerprints)
+				clusterClient, err := proxmox.NewClusterClientWithContext(ctx, pve.Name, clientConfig, endpoints, endpointFingerprints, conflictingEndpoints...)
 				if ctx.Err() != nil {
 					return
 				}

internal/monitoring/monitor_pve_client_endpoints.go

@@ -1,15 +1,17 @@
 package monitoring
 
 import (
+	"strings"
+
 	"github.com/rcourtman/pulse-go-rewrite/internal/config"
 	"github.com/rs/zerolog/log"
 )
 
-func (m *Monitor) buildClusterEndpointsForInit(pve config.PVEInstance) ([]string, map[string]string) {
+func (m *Monitor) buildClusterEndpointsForInit(pve config.PVEInstance) ([]string, map[string]string, []string) {
 	return m.buildClusterClientEndpoints(pve)
 }
 
-func (m *Monitor) buildClusterEndpointsForReconnect(pve config.PVEInstance) ([]string, map[string]string) {
+func (m *Monitor) buildClusterEndpointsForReconnect(pve config.PVEInstance) ([]string, map[string]string, []string) {
 	return m.buildClusterClientEndpoints(pve)
 }
 
@@ -18,9 +20,11 @@ func (m *Monitor) buildClusterEndpointsForReconnect(pve config.PVEInstance) ([]s
 // are valid inside the cluster but unreachable from the Pulse container; those
 // addresses remain useful failover and reachability evidence, but they must
 // never displace the URL the operator proved reachable when adding the cluster.
-func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string, map[string]string) {
+func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string, map[string]string, []string) {
 	endpoints := make([]string, 0, len(pve.ClusterEndpoints)+1)
 	endpointFingerprints := make(map[string]string)
+	var conflictingEndpoints []string
+	conflicts := make(map[string]bool)
 	discoveryCfg := monitorDiscoveryConfig(m)
 	hasFingerprint := pve.Fingerprint != ""
 	seen := make(map[string]struct{}, len(pve.ClusterEndpoints)+1)
@@ -53,7 +57,17 @@ func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string,
 			failoverCount++
 		}
 		if ep.Fingerprint != "" {
-			endpointFingerprints[host] = ep.Fingerprint
+			if previous, exists := endpointFingerprints[host]; exists {
+				if strings.ToLower(strings.ReplaceAll(previous, ":", "")) != strings.ToLower(strings.ReplaceAll(ep.Fingerprint, ":", "")) && !conflicts[host] {
+					// The map cannot retain two values for the exact same URL.
+					// Carry the conflict separately; a later repeated row must
+					// not erase it or choose the certificate to trust.
+					conflicts[host] = true
+					conflictingEndpoints = append(conflictingEndpoints, host)
+				}
+			} else {
+				endpointFingerprints[host] = ep.Fingerprint
+			}
 		}
 	}
 
@@ -71,5 +85,5 @@ func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string,
 			Msg("Configured Proxmox cluster URL is empty; using first discovered member as API authority")
 	}
 
-	return endpoints, endpointFingerprints
+	return endpoints, endpointFingerprints, conflictingEndpoints
 }

internal/monitoring/monitor_pve_cluster.go

@@ -514,6 +514,16 @@ func (m *Monitor) detectClusterMembership(ctx context.Context, instanceName stri
 // node addresses changed, the config is persisted and the failover client is
 // rebuilt so polling stops dialing the dead addresses.
 func (m *Monitor) refreshClusterEndpoints(instanceName string, instanceCfg *config.PVEInstance, clusterName string, discovered []config.ClusterEndpoint) {
+	// Membership discovery is not a trust decision. In particular it can
+	// collapse duplicate saved rows, or move an alias away from its conflicting
+	// authority. Keep the saved evidence and current sibling clients until an
+	// operator independently reconciles trust through normal configuration.
+	_, savedPins, conflicts := m.buildClusterClientEndpoints(*instanceCfg)
+	if err := proxmox.CheckEndpointFingerprintConflicts(savedPins, conflicts...); err != nil {
+		log.Warn().Str("instance", instanceName).Err(err).
+			Msg("Retaining cluster endpoints until conflicting saved TLS trust is reconciled")
+		return
+	}
 	refreshed := mergeRefreshedClusterEndpoints(instanceCfg.ClusterEndpoints, discovered, instanceCfg.VerifySSL)
 	if !clusterEndpointIdentityChanged(instanceCfg.ClusterEndpoints, refreshed) {
 		return
@@ -700,10 +710,10 @@ func (m *Monitor) rebuildPVEClusterClient(instanceName string) {
 		return
 	}
 
-	endpoints, endpointFingerprints := m.buildClusterEndpointsForReconnect(*pve)
+	endpoints, endpointFingerprints, conflictingEndpoints := m.buildClusterEndpointsForReconnect(*pve)
 	clientConfig := config.CreateProxmoxConfig(pve)
 	clientConfig.Timeout = m.config.ConnectionTimeout
-	clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints)
+	clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints, conflictingEndpoints...)
 
 	m.replacePVEClient(instanceName, clusterClient)
 

internal/monitoring/monitor_pve_cluster_refresh_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -362,13 +363,16 @@ func TestBuildClusterEndpoints_PreservesConfiguredAuthority(t *testing.T) {
 
 	for _, build := range []struct {
 		name string
-		fn   func(config.PVEInstance) ([]string, map[string]string)
+		fn   func(config.PVEInstance) ([]string, map[string]string, []string)
 	}{
 		{name: "initialization", fn: monitor.buildClusterEndpointsForInit},
 		{name: "reconnect", fn: monitor.buildClusterEndpointsForReconnect},
 	} {
 		t.Run(build.name, func(t *testing.T) {
-			endpoints, fingerprints := build.fn(instance)
+			endpoints, fingerprints, conflicts := build.fn(instance)
+			if len(conflicts) != 0 {
+				t.Fatalf("non-conflicting member trust was refused: %v", conflicts)
+			}
 			if len(endpoints) != len(want) {
 				t.Fatalf("expected %d ordered endpoints, got %v", len(want), endpoints)
 			}
@@ -422,3 +426,49 @@ func TestStartupClientConstructionHonoursCancellation(t *testing.T) {
 func TestPBSStartupCancellationKeepsFirstPollBoundary(t *testing.T) {
 	testPBSStartupCancellationKeepsFirstPollBoundary(t)
 }
+
+func TestBuildClusterEndpointsRetainsDuplicatePinConflicts(t *testing.T) {
+	monitor := &Monitor{config: &config.Config{}}
+	const member = "https://member.example.invalid:8006"
+	for _, build := range []struct {
+		name string
+		fn   func(config.PVEInstance) ([]string, map[string]string, []string)
+	}{
+		{"init", monitor.buildClusterEndpointsForInit}, {"reconnect", monitor.buildClusterEndpointsForReconnect},
+	} {
+		for _, sequence := range [][]string{{"aa:bb", "AA:BB", "aabb"}, {"aa:bb", "cc:dd"}, {"cc:dd", "aa:bb", "cc:dd"}, {"aa:bb", "", "cc:dd", "aa:bb"}, {"", "aa:bb", "cc:dd"}, {"", "aa:bb", "AA:BB"}, {":", ":"}} {
+			t.Run(build.name+"/"+strings.Join(sequence, "_"), func(t *testing.T) {
+				inst := config.PVEInstance{Host: "https://primary.example.invalid:8006", VerifySSL: true}
+				for _, pin := range sequence {
+					inst.ClusterEndpoints = append(inst.ClusterEndpoints, config.ClusterEndpoint{NodeName: "member", Host: member, Fingerprint: pin})
+				}
+				endpoints, pins, conflicts := build.fn(inst)
+				wantConflict := false
+				firstPin := ""
+				for _, pin := range sequence {
+					if firstPin == "" {
+						firstPin = pin
+					}
+					if pin != "" && strings.ToLower(strings.ReplaceAll(pin, ":", "")) != strings.ToLower(strings.ReplaceAll(firstPin, ":", "")) {
+						wantConflict = true
+					}
+				}
+				if len(endpoints) != 2 || pins[member] != firstPin {
+					t.Fatal("duplicate rows changed routing order or selected later trust")
+				}
+				if wantConflict {
+					if len(conflicts) != 1 || conflicts[0] != member {
+						t.Fatal("later row erased conflicting certificate evidence")
+					}
+				} else if len(conflicts) != 0 {
+					t.Fatal("agreeing/absent member pins created a conflict")
+				}
+				for i, ep := range inst.ClusterEndpoints {
+					if ep.Fingerprint != sequence[i] || ep.Host != member {
+						t.Fatal("producer changed saved configuration")
+					}
+				}
+			})
+		}
+	}
+}

internal/monitoring/monitor_pve_pin_conflict_test.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"reflect"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -25,110 +26,235 @@ import (
 // deployment/recovery, and says nothing about any reported incident's cause.
 func TestClusterConflictingPinsEndpointProjection(t *testing.T) {
 	for _, mode := range []string{"token", "password-json", "password-form"} {
-		t.Run(mode, func(t *testing.T) {
-			var badRequests atomic.Int32
-			endpoint := func(bad bool) *httptest.Server {
-				s := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-					if bad {
-						badRequests.Add(1)
-					}
-					switch r.URL.Path {
-					case "/api2/json/access/ticket":
-						if mode == "password-form" && r.Header.Get("Content-Type") == "application/json" {
-							w.WriteHeader(http.StatusUnsupportedMediaType)
-							fmt.Fprint(w, `{"message":"fixture requires form encoding"}`)
-							return
+		for _, layout := range []string{"url-aliases", "exact-duplicates"} {
+			t.Run(layout+"/"+mode, func(t *testing.T) {
+				var badRequests atomic.Int32
+				endpoint := func(bad bool) *httptest.Server {
+					s := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+						if bad {
+							badRequests.Add(1)
 						}
-						fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket","CSRFPreventionToken":"fixture-csrf"}}`)
-					case "/api2/json/nodes":
-						if mode == "token" {
-							if r.Header.Get("Authorization") != "PVEAPIToken=fixture@pve!pulse=fixture" {
-								t.Error("lost token fixture auth")
+						switch r.URL.Path {
+						case "/api2/json/access/ticket":
+							if mode == "password-form" && r.Header.Get("Content-Type") == "application/json" {
+								w.WriteHeader(http.StatusUnsupportedMediaType)
+								fmt.Fprint(w, `{"message":"fixture requires form encoding"}`)
+								return
 							}
-						} else if r.Header.Get("Cookie") != "PVEAuthCookie=fixture-ticket" {
-							t.Error("lost fixture session")
+							fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket","CSRFPreventionToken":"fixture-csrf"}}`)
+						case "/api2/json/nodes":
+							if mode == "token" {
+								if r.Header.Get("Authorization") != "PVEAPIToken=fixture@pve!pulse=fixture" {
+									t.Error("lost token fixture auth")
+								}
+							} else if r.Header.Get("Cookie") != "PVEAuthCookie=fixture-ticket" {
+								t.Error("lost fixture session")
+							}
+							fmt.Fprint(w, `{"data":[{"node":"primary","status":"online","cpu":0.125},{"node":"member","status":"online","cpu":0.25},{"node":"member-alias","status":"online","cpu":0.25}]}`)
+						default:
+							t.Errorf("left ordinary login/nodes path: %s", r.URL.Path)
+							http.NotFound(w, r)
+						}
+					}))
+					t.Cleanup(s.Close)
+					return s
+				}
+				primary, member := endpoint(false), endpoint(true)
+				pin := func(s *httptest.Server) string {
+					sum := sha256.Sum256(s.Certificate().Raw)
+					return hex.EncodeToString(sum[:])
+				}
+				inst := config.PVEInstance{Name: "pin-conflict-cluster", Host: primary.URL, IsCluster: true, ClusterName: "Fixture cluster", VerifySSL: true, Fingerprint: pin(primary), ClusterEndpoints: []config.ClusterEndpoint{
+					{NodeName: "primary", NodeIdentity: "primary-identity", Host: primary.URL, Fingerprint: pin(primary)},
+					{NodeName: "member", NodeIdentity: "member-identity", Host: member.URL, Fingerprint: pin(member)},
+					{NodeName: "member-alias", NodeIdentity: "alias-identity", Host: member.URL + "/./", Fingerprint: strings.Repeat("0", 64)},
+				}}
+				if layout == "exact-duplicates" {
+					inst.ClusterEndpoints[2].Host = member.URL
+				}
+				m := &Monitor{config: &config.Config{PVEInstances: []config.PVEInstance{inst}}}
+				construct := func(conflict bool) *proxmox.ClusterClient {
+					endpoints, pins, conflicts := m.buildClusterEndpointsForInit(inst)
+					wantEndpoints, wantConflicts := 3, 0
+					if layout == "exact-duplicates" {
+						wantEndpoints = 2
+						if conflict {
+							wantConflicts = 1
 						}
-						fmt.Fprint(w, `{"data":[{"node":"primary","status":"online","cpu":0.125},{"node":"member","status":"online","cpu":0.25},{"node":"member-alias","status":"online","cpu":0.25}]}`)
-					default:
-						t.Errorf("left ordinary login/nodes path: %s", r.URL.Path)
-						http.NotFound(w, r)
 					}
-				}))
-				t.Cleanup(s.Close)
-				return s
-			}
-			primary, member := endpoint(false), endpoint(true)
-			pin := func(s *httptest.Server) string {
-				sum := sha256.Sum256(s.Certificate().Raw)
-				return hex.EncodeToString(sum[:])
-			}
-			inst := config.PVEInstance{Name: "pin-conflict-cluster", Host: primary.URL, IsCluster: true, ClusterName: "Fixture cluster", VerifySSL: true, Fingerprint: pin(primary), ClusterEndpoints: []config.ClusterEndpoint{
-				{NodeName: "primary", NodeIdentity: "primary-identity", Host: primary.URL, Fingerprint: pin(primary)},
-				{NodeName: "member", NodeIdentity: "member-identity", Host: member.URL, Fingerprint: pin(member)},
-				{NodeName: "member-alias", NodeIdentity: "alias-identity", Host: member.URL + "/./", Fingerprint: strings.Repeat("0", 64)},
-			}}
-			m := &Monitor{config: &config.Config{PVEInstances: []config.PVEInstance{inst}}}
-			construct := func() *proxmox.ClusterClient {
-				endpoints, pins := m.buildClusterEndpointsForInit(inst)
-				if len(endpoints) != 3 || len(pins) != 3 {
-					t.Fatalf("saved alias evidence was collapsed: endpoints=%d pins=%d", len(endpoints), len(pins))
-				}
-				cfg := proxmox.ClientConfig{Host: inst.Host, Fingerprint: inst.Fingerprint, VerifySSL: true, Timeout: 2 * time.Second}
-				if mode == "token" {
-					cfg.TokenName = "fixture@pve!pulse"
-					cfg.TokenValue = "fixture"
-				} else {
-					cfg.User = "fixture@pam"
-					cfg.Password = "fixture-password"
-				}
-				return proxmox.NewClusterClient(inst.Name, cfg, endpoints, pins)
-			}
-			project := func(cc *proxmox.ClusterClient, phase string, conflict bool) {
-				nodes, err := cc.GetNodes(context.Background())
-				if err != nil || len(nodes) != 3 {
-					t.Fatalf("trusted primary lost ordinary monitoring: %v", err)
-				}
-				model := make([]models.Node, len(nodes))
-				for i, node := range nodes {
-					model[i] = models.Node{Name: node.Node, Status: node.Status}
-				}
-				m.updateClusterEndpointStatus(inst.Name, &inst, cc, model)
-				for i, ep := range inst.ClusterEndpoints {
-					if !ep.Online || ep.NodeIdentity != []string{"primary-identity", "member-identity", "alias-identity"}[i] {
-						t.Fatal("trust refusal changed native online state or identity")
+					if len(endpoints) != wantEndpoints || len(pins) != wantEndpoints || len(conflicts) != wantConflicts {
+						t.Fatalf("saved alias evidence was collapsed: endpoints=%d pins=%d", len(endpoints), len(pins))
 					}
-					if ep.PulseReachable == nil || *ep.PulseReachable != (i == 0 || !conflict) || ep.LastPulseCheck == nil {
-						t.Fatalf("missing/false Pulse reachability observation: %+v", ep)
-					}
-					if i > 0 && conflict && !strings.Contains(ep.PulseError, "TLS trust conflict") {
-						t.Fatalf("member lost actionable trust diagnostic: %+v", ep)
-					}
-					if !conflict && ep.PulseError != "" {
-						t.Fatalf("agreed trust kept a stale failure: %+v", ep)
+					cfg := proxmox.ClientConfig{Host: inst.Host, Fingerprint: inst.Fingerprint, VerifySSL: true, Timeout: 2 * time.Second}
+					if mode == "token" {
+						cfg.TokenName = "fixture@pve!pulse"
+						cfg.TokenValue = "fixture"
+					} else {
+						cfg.User = "fixture@pam"
+						cfg.Password = "fixture-password"
 					}
+					return proxmox.NewClusterClient(inst.Name, cfg, endpoints, pins, conflicts...)
 				}
-				wire, err := json.Marshal(struct {
-					Case, Phase string
-					Endpoints   []config.ClusterEndpoint
-				}{mode, phase, inst.ClusterEndpoints})
-				if err != nil {
-					t.Fatal(err)
+				project := func(cc *proxmox.ClusterClient, phase string, conflict bool) {
+					nodes, err := cc.GetNodes(context.Background())
+					if err != nil || len(nodes) != 3 {
+						t.Fatalf("trusted primary lost ordinary monitoring: %v", err)
+					}
+					model := make([]models.Node, len(nodes))
+					for i, node := range nodes {
+						model[i] = models.Node{Name: node.Node, Status: node.Status}
+					}
+					m.updateClusterEndpointStatus(inst.Name, &inst, cc, model)
+					for i, ep := range inst.ClusterEndpoints {
+						if !ep.Online || ep.NodeIdentity != []string{"primary-identity", "member-identity", "alias-identity"}[i] {
+							t.Fatal("trust refusal changed native online state or identity")
+						}
+						if ep.PulseReachable == nil || *ep.PulseReachable != (i == 0 || !conflict) || ep.LastPulseCheck == nil {
+							t.Fatalf("missing/false Pulse reachability observation: %+v", ep)
+						}
+						if i > 0 && conflict && !strings.Contains(ep.PulseError, "TLS trust conflict") {
+							t.Fatalf("member lost actionable trust diagnostic: %+v", ep)
+						}
+						if !conflict && ep.PulseError != "" {
+							t.Fatalf("agreed trust kept a stale failure: %+v", ep)
+						}
+					}
+					wire, err := json.Marshal(struct {
+						Case, Phase string
+						Endpoints   []config.ClusterEndpoint
+					}{layout + "/" + mode, phase, inst.ClusterEndpoints})
+					if err != nil {
+						t.Fatal(err)
+					}
+					t.Logf("PIN_CONFLICT_ENDPOINT_PROJECTION_JSON=%s", wire)
 				}
-				t.Logf("PIN_CONFLICT_ENDPOINT_PROJECTION_JSON=%s", wire)
-			}
-			refused := construct()
-			project(refused, "conflicting-saved-trust", true)
-			if badRequests.Load() != 0 || inst.ClusterEndpoints[2].Fingerprint != strings.Repeat("0", 64) {
-				t.Fatal("conflict reached credentials or rewrote saved trust")
-			}
-			// Normal source configuration reload, not mutation of the owned map or a
-			// forced/favourable repeat on the refused source. Native state stays online.
-			inst.ClusterEndpoints[2].Fingerprint = pin(member)
-			accepted := construct()
-			project(accepted, "reconciled-config-reload", false)
-			if badRequests.Load() == 0 {
-				t.Fatal("agreed saved trust never reached accepted ordinary use")
-			}
-		})
+				refused := construct(true)
+				project(refused, "conflicting-saved-trust", true)
+				if badRequests.Load() != 0 || inst.ClusterEndpoints[2].Fingerprint != strings.Repeat("0", 64) {
+					t.Fatal("conflict reached credentials or rewrote saved trust")
+				}
+				// Normal source configuration reload, not mutation of the owned map or a
+				// forced/favourable repeat on the refused source. Native state stays online.
+				inst.ClusterEndpoints[2].Fingerprint = pin(member)
+				accepted := construct(false)
+				project(accepted, "reconciled-config-reload", false)
+				if badRequests.Load() == 0 {
+					t.Fatal("agreed saved trust never reached accepted ordinary use")
+				}
+			})
+		}
+	}
+}
+
+// Exercise the owning monitor routes, not just equivalent calls to their
+// builder. Discovery is allowed to report a collapsed/changed member list,
+// but must not make saved ambiguity disappear or interrupt trusted siblings.
+func TestClusterPinConflictsSurviveMonitorLifecycle(t *testing.T) {
+	originalDelays := connRetryDelays
+	connRetryDelays = []time.Duration{time.Millisecond, time.Millisecond}
+	t.Cleanup(func() { connRetryDelays = originalDelays })
+	for _, mode := range []string{"token", "password-json", "password-form"} {
+		for _, layout := range []string{"url-aliases", "exact-duplicates"} {
+			for _, route := range []string{"init", "reconnect", "refresh"} {
+				t.Run(layout+"/"+mode+"/"+route, func(t *testing.T) {
+					var badRequests, safeReads atomic.Int32
+					endpoint := func(bad bool) *httptest.Server {
+						s := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+							if bad {
+								badRequests.Add(1)
+							}
+							switch r.URL.Path {
+							case "/api2/json/access/ticket":
+								if mode == "password-form" && r.Header.Get("Content-Type") == "application/json" {
+									w.WriteHeader(http.StatusUnsupportedMediaType)
+									return
+								}
+								fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket","CSRFPreventionToken":"fixture-csrf"}}`)
+							case "/api2/json/nodes":
+								if !bad {
+									safeReads.Add(1)
+								}
+								fmt.Fprint(w, `{"data":[{"node":"primary","status":"online"},{"node":"member","status":"online"}]}`)
+							default:
+								t.Errorf("unexpected fixture request: %s", r.URL.Path)
+								http.NotFound(w, r)
+							}
+						}))
+						t.Cleanup(s.Close)
+						return s
+					}
+					primary, member := endpoint(false), endpoint(true)
+					pin := func(s *httptest.Server) string {
+						sum := sha256.Sum256(s.Certificate().Raw)
+						return hex.EncodeToString(sum[:])
+					}
+					alias := member.URL + "/./"
+					if layout == "exact-duplicates" {
+						alias = member.URL
+					}
+					inst := config.PVEInstance{Name: "lifecycle-cluster", Host: primary.URL, IsCluster: true, ClusterName: "Fixture cluster", VerifySSL: true, Fingerprint: pin(primary), ClusterEndpoints: []config.ClusterEndpoint{
+						{NodeName: "primary", NodeIdentity: "primary-id", Host: primary.URL, Fingerprint: pin(primary)},
+						{NodeName: "member", NodeIdentity: "member-id", Host: member.URL, Fingerprint: pin(member)},
+						{NodeName: "member-alias", NodeIdentity: "alias-id", Host: alias, Fingerprint: strings.Repeat("0", 64)},
+					}}
+					if mode == "token" {
+						inst.TokenName, inst.TokenValue = "fixture@pve!pulse", "fixture"
+					} else {
+						inst.User, inst.Password = "fixture@pam", "fixture-password"
+					}
+					m := &Monitor{config: &config.Config{PVEInstances: []config.PVEInstance{inst}, ConnectionTimeout: 2 * time.Second}, pveClients: make(map[string]PVEClientInterface)}
+					switch route {
+					case "init":
+						m.initPVEClients(m.config)
+					case "reconnect":
+						ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+						defer cancel()
+						m.retryFailedConnections(ctx)
+						if err := ctx.Err(); err != nil {
+							t.Fatalf("missing-client retry did not finish: %v", err)
+						}
+					case "refresh":
+						m.rebuildPVEClusterClient(inst.Name)
+						original := m.pveClients[inst.Name]
+						// Discovery both drops the contradictory row and changes a
+						// sibling's address. Neither is permission to reconcile trust.
+						discovered := append([]config.ClusterEndpoint(nil), inst.ClusterEndpoints[:2]...)
+						discovered[0].Host += "/"
+						m.refreshClusterEndpoints(inst.Name, &m.config.PVEInstances[0], inst.ClusterName, discovered)
+						if m.pveClients[inst.Name] != original || !reflect.DeepEqual(m.config.PVEInstances[0], inst) {
+							t.Fatal("discovery erased conflicting saved trust or replaced active sibling clients")
+						}
+					}
+					cc, ok := m.pveClients[inst.Name].(*proxmox.ClusterClient)
+					if !ok {
+						t.Fatal("owning monitor route did not retain its cluster client")
+					}
+					if _, err := cc.GetNodes(context.Background()); err != nil || safeReads.Load() == 0 || badRequests.Load() != 0 {
+						t.Fatalf("monitor route lost sibling use or dispatched conflicting trust: %v safe=%d bad=%d", err, safeReads.Load(), badRequests.Load())
+					}
+					if health := cc.GetHealthStatusWithErrors()[member.URL]; health.Healthy || !strings.Contains(health.LastError, "TLS trust conflict") {
+						t.Fatalf("monitor route lost member's trust refusal: %+v", health)
+					}
+					if !reflect.DeepEqual(m.config.PVEInstances[0], inst) {
+						t.Fatal("monitor route changed saved configuration")
+					}
+					if route == "refresh" {
+						// A normal independently reconciled configuration removes
+						// the hold; automatic address refresh is not disabled forever.
+						m.config.PVEInstances[0].ClusterEndpoints[2].Fingerprint = pin(member)
+						discovered := append([]config.ClusterEndpoint(nil), m.config.PVEInstances[0].ClusterEndpoints...)
+						discovered[0].Host += "/"
+						m.refreshClusterEndpoints(inst.Name, &m.config.PVEInstances[0], inst.ClusterName, discovered)
+						reloaded := m.pveClients[inst.Name]
+						if reloaded == cc {
+							t.Fatal("reconciled configuration did not resume address refresh")
+						}
+						if _, err := reloaded.GetNodes(context.Background()); err != nil || badRequests.Load() == 0 {
+							t.Fatalf("reconciled refresh did not resume ordinary member use: %v", err)
+						}
+					}
+				})
+			}
+		}
 	}
 }

pkg/proxmox/cluster_client.go

@@ -20,18 +20,19 @@ import (
 
 // ClusterClient wraps multiple Proxmox clients for cluster-aware operations
 type ClusterClient struct {
-	mu                   sync.RWMutex
-	clientInitMu         sync.Mutex // Serialize lazy construction while a trust probe releases mu.
-	name                 string
-	clients              map[string]*Client   // Key is node name
-	endpoints            []string             // All available endpoints
-	endpointFingerprints map[string]string    // Per-endpoint TLS fingerprints (TOFU)
-	nodeHealth           map[string]bool      // Track node health
-	lastHealthCheck      map[string]time.Time // Track last health check time
-	lastError            map[string]string    // Track last error per endpoint
-	config               ClientConfig         // Base config (auth info)
-	rateLimitUntil       map[string]time.Time // Cooldown window for rate-limited endpoints
-	recoveryInProgress   bool                 // Prevent overlapping background recovery sweeps
+	mu                              sync.RWMutex
+	clientInitMu                    sync.Mutex // Serialize lazy construction while a trust probe releases mu.
+	name                            string
+	clients                         map[string]*Client   // Key is node name
+	endpoints                       []string             // All available endpoints
+	endpointFingerprints            map[string]string    // Per-endpoint TLS fingerprints (TOFU)
+	conflictingEndpointFingerprints map[string]bool      // Exact duplicate saved rows refused by the producer.
+	nodeHealth                      map[string]bool      // Track node health
+	lastHealthCheck                 map[string]time.Time // Track last health check time
+	lastError                       map[string]string    // Track last error per endpoint
+	config                          ClientConfig         // Base config (auth info)
+	rateLimitUntil                  map[string]time.Time // Cooldown window for rate-limited endpoints
+	recoveryInProgress              bool                 // Prevent overlapping background recovery sweeps
 }
 
 const (
@@ -157,15 +158,17 @@ func sanitizeEndpointError(errMsg string) string {
 // NewClusterClient creates a new cluster-aware client.
 // endpointFingerprints is an optional map of endpoint URL -> TLS fingerprint for per-node certificate verification.
 // This enables TOFU (Trust On First Use) for clusters with unique self-signed certs per node.
-func NewClusterClient(name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string) *ClusterClient {
-	cc, _ := NewClusterClientWithContext(context.Background(), name, config, endpoints, endpointFingerprints)
+// Optional conflictingEndpoints retain contradictory exact duplicate saved rows
+// which a map cannot represent. They refuse only those endpoint trust origins.
+func NewClusterClient(name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string, conflictingEndpoints ...string) *ClusterClient {
+	cc, _ := NewClusterClientWithContext(context.Background(), name, config, endpoints, endpointFingerprints, conflictingEndpoints...)
 	return cc
 }
 
 // NewClusterClientWithContext keeps synchronous startup checks and password
 // session construction within the caller's lifetime. Endpoint failures still
 // produce a recoverable cluster; cancellation does not publish a partial one.
-func NewClusterClientWithContext(ctx context.Context, name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string) (*ClusterClient, error) {
+func NewClusterClientWithContext(ctx context.Context, name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string, conflictingEndpoints ...string) (*ClusterClient, error) {
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
@@ -180,16 +183,21 @@ func NewClusterClientWithContext(ctx context.Context, name string, config Client
 	for endpoint, fingerprint := range endpointFingerprints {
 		ownedFingerprints[endpoint] = fingerprint
 	}
+	ownedConflicts := make(map[string]bool, len(conflictingEndpoints))
+	for _, endpoint := range conflictingEndpoints {
+		ownedConflicts[endpoint] = true
+	}
 	cc := &ClusterClient{
-		name:                 name,
-		clients:              make(map[string]*Client),
-		endpoints:            endpoints,
-		endpointFingerprints: ownedFingerprints,
-		nodeHealth:           make(map[string]bool),
-		lastHealthCheck:      make(map[string]time.Time),
-		lastError:            make(map[string]string),
-		config:               config,
-		rateLimitUntil:       make(map[string]time.Time),
+		name:                            name,
+		clients:                         make(map[string]*Client),
+		endpoints:                       endpoints,
+		endpointFingerprints:            ownedFingerprints,
+		conflictingEndpointFingerprints: ownedConflicts,
+		nodeHealth:                      make(map[string]bool),
+		lastHealthCheck:                 make(map[string]time.Time),
+		lastError:                       make(map[string]string),
+		config:                          config,
+		rateLimitUntil:                  make(map[string]time.Time),
 	}
 
 	// Initialize all endpoints as unknown (will be tested on first use)
@@ -240,6 +248,12 @@ func clusterEndpointKey(endpoint string) (string, bool) {
 
 func (cc *ClusterClient) knownEndpointFingerprintLocked(endpoint string) (string, bool, error) {
 	key, valid := clusterEndpointKey(endpoint)
+	for conflict := range cc.conflictingEndpointFingerprints {
+		conflictKey, conflictValid := clusterEndpointKey(conflict)
+		if conflict == endpoint || (valid && conflictValid && conflictKey == key) {
+			return "", true, errConflictingEndpointFingerprints
+		}
+	}
 	var fingerprint string
 	known := false
 	for storedEndpoint, fp := range cc.endpointFingerprints {
@@ -268,6 +282,24 @@ func (cc *ClusterClient) knownEndpointFingerprintLocked(endpoint string) (string
 	return fingerprint, known, nil
 }
 
+// CheckEndpointFingerprintConflicts checks saved trust without constructing a
+// client, probing TLS or changing any pin. Discovery must not discard ambiguity
+// by collapsing saved rows or replacing their addresses during a refresh.
+// This checks agreement only; malformed nonempty pins remain the TLS verifier's
+// responsibility and are never treated as absent trust here.
+func CheckEndpointFingerprintConflicts(fingerprints map[string]string, conflictingEndpoints ...string) error {
+	if len(conflictingEndpoints) > 0 {
+		return errConflictingEndpointFingerprints
+	}
+	cc := ClusterClient{endpointFingerprints: fingerprints}
+	for endpoint := range fingerprints {
+		if _, _, err := cc.knownEndpointFingerprintLocked(endpoint); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // Only a declared, previously untrusted member may use the primary pin as a
 // first-use bootstrap. A known member or the configured primary must keep its
 // established pin; recovery is never permission to replace it.

pkg/proxmox/cluster_client_api_test.go

@@ -930,7 +930,11 @@ func TestClusterSavedPinResolutionRejectsConflicts(t *testing.T) {
 		{"member.example.invalid:8006/tenant/./", "https://member.example.invalid:8006/tenant"},
 	}
 	for _, pair := range pairs {
-		cc := NewClusterClient("resolution", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin"}, nil, map[string]string{pair[0]: "saved-pin", pair[1]: "other-pin"})
+		pins := map[string]string{pair[0]: "saved-pin", pair[1]: "other-pin"}
+		if err := CheckEndpointFingerprintConflicts(pins); !errors.Is(err, errConflictingEndpointFingerprints) {
+			t.Errorf("refresh check lost contradictory saved trust: %v", err)
+		}
+		cc := NewClusterClient("resolution", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin"}, nil, pins)
 		for _, endpoint := range pair {
 			fingerprint, err := cc.getEndpointFingerprint(endpoint)
 			if fingerprint != "" || !errors.Is(err, errConflictingEndpointFingerprints) {
@@ -948,7 +952,11 @@ func TestClusterSavedPinResolutionRejectsConflicts(t *testing.T) {
 	// Invalid nonempty pin presentation cannot turn into unpinned CA/opt-out
 	// admission, even when every saved spelling agrees after normalization.
 	for _, pin := range []string{":", "::", " ", "not-a-pin"} {
-		cc := NewClusterClient("invalid-pin", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin", VerifySSL: false}, nil, map[string]string{"https://member.example.invalid:443/": pin, "https://MEMBER.example.invalid": pin})
+		pins := map[string]string{"https://member.example.invalid:443/": pin, "https://MEMBER.example.invalid": pin}
+		if err := CheckEndpointFingerprintConflicts(pins); err != nil {
+			t.Errorf("refresh check refused agreeing nonempty pins: %v", err)
+		}
+		cc := NewClusterClient("invalid-pin", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin", VerifySSL: false}, nil, pins)
 		got, err := cc.getEndpointFingerprint("https://member.example.invalid")
 		if err != nil || got != pin || got == "" {
 			t.Errorf("invalid saved pin became absent trust: %q %v", got, err)
@@ -991,3 +999,25 @@ func TestClusterFirstUseCaptureDoesNotReconcileConflictingPins(t *testing.T) {
 		t.Error("capture authenticated or reconciled contradictory saved trust")
 	}
 }
+
+func TestClusterProducerPinConflictsAreOwnedAndScoped(t *testing.T) {
+	const original = "https://member.example.invalid:443/tenant/"
+	input := []string{original}
+	if err := CheckEndpointFingerprintConflicts(map[string]string{original: "saved-pin"}, input...); !errors.Is(err, errConflictingEndpointFingerprints) {
+		t.Errorf("refresh check discarded producer's duplicate-row conflict: %v", err)
+	}
+	cc := NewClusterClient("producer-conflict", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "primary-pin"}, nil, map[string]string{original: "saved-pin"}, input...)
+	input[0] = "https://other.example.invalid/tenant"
+	for _, endpoint := range []string{original, "https://MEMBER.example.invalid/tenant"} {
+		fingerprint, err := cc.getEndpointFingerprint(endpoint)
+		if fingerprint != "" || !errors.Is(err, errConflictingEndpointFingerprints) {
+			t.Errorf("producer's copied conflict was lost: %q %v", fingerprint, err)
+		}
+	}
+	for _, endpoint := range []string{"https://other.example.invalid/tenant", "https://member.example.invalid/other", "http://member.example.invalid/tenant", "https://member.example.invalid:8006/tenant"} {
+		fingerprint, err := cc.getEndpointFingerprint(endpoint)
+		if err != nil || fingerprint != "primary-pin" {
+			t.Errorf("producer conflict refused an independent authority/path: %q %v", fingerprint, err)
+		}
+	}
+}

pkg/proxmox/cluster_pin_conflict_test.go

@@ -169,7 +169,11 @@ func TestClusterSavedPinScopesStayDistinct(t *testing.T) {
 	const endpoint = "https://member.scope.invalid:8006/tenant"
 	const pin = "saved-pin"
 	for _, other := range []string{"http://member.scope.invalid:8006/tenant", "https://member.scope.invalid:8007/tenant", "https://other.scope.invalid:8006/tenant", "https://member.scope.invalid:8006/Tenant", "https://member.scope.invalid:8006/other"} {
-		cc := NewClusterClient("scope", ClientConfig{Host: endpoint, Fingerprint: "primary-pin"}, nil, map[string]string{endpoint: pin, other: "independent-pin"})
+		pins := map[string]string{endpoint: pin, other: "independent-pin"}
+		if err := CheckEndpointFingerprintConflicts(pins); err != nil {
+			t.Errorf("independent origin/base path held refresh: %v", err)
+		}
+		cc := NewClusterClient("scope", ClientConfig{Host: endpoint, Fingerprint: "primary-pin"}, nil, pins)
 		fingerprint, err := cc.getEndpointFingerprint(endpoint)
 		if err != nil || fingerprint != pin {
 			t.Errorf("distinct endpoint created a saved-trust conflict: %q %v", fingerprint, err)