Fix fresh cluster bootstrap deadlock and stale peer discovery

cmd/flux-agent/bootstrap_identity.go

@@ -54,7 +54,7 @@ func localNodeIdentity(cfg *config.Config) nodeIdentity {
 		PatroniScope:   cfg.PatroniScope,
 		NodeName:       cfg.MyName,
 		PGDataEmpty:    directoryIsEmpty("/var/lib/postgresql/data"),
-		EtcdDataEmpty:  directoryIsEmpty("/var/lib/etcd"),
+		EtcdDataEmpty:  etcdDataDirectoryIsFresh("/var/lib/etcd"),
 		MembershipView: membershipNames(cfg.EtcdInitialCluster),
 	}
 	if !identity.PGDataEmpty {
@@ -88,6 +88,27 @@ func directoryIsEmpty(path string) bool {
 	return err == nil && len(entries) == 0
 }
 
+// etcdDataDirectoryIsFresh applies the same strict empty-directory policy as
+// directoryIsEmpty, except for the agent-owned .etcd3_api breadcrumb. The
+// marker is written before execing etcd and is not etcd state; treating it as
+// preserved state prevents an otherwise fresh node from ever retrying a safe
+// bootstrap after an unsuccessful etcd launch.
+func etcdDataDirectoryIsFresh(path string) bool {
+	entries, err := os.ReadDir(path)
+	if os.IsNotExist(err) {
+		return true
+	}
+	if err != nil {
+		return false
+	}
+	for _, entry := range entries {
+		if entry.Name() != ".etcd3_api" || !entry.Type().IsRegular() {
+			return false
+		}
+	}
+	return true
+}
+
 // identityProbeToken authenticates the read-only peer probe without exposing
 // the replication password itself. The endpoint still returns no secrets, but
 // requiring a shared HMAC prevents arbitrary Internet clients from inventorying

cmd/flux-agent/bootstrap_identity_test.go

@@ -153,6 +153,34 @@ func TestDirectoryIsEmptyIsStrict(t *testing.T) {
 	}
 }
 
+func TestEtcdDataDirectoryIsFreshAllowsOnlyAgentMarker(t *testing.T) {
+	dir := t.TempDir()
+	marker := filepath.Join(dir, ".etcd3_api")
+	if err := os.WriteFile(marker, []byte("etcd3\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if !etcdDataDirectoryIsFresh(dir) {
+		t.Fatal("the agent-owned marker must not count as persisted etcd state")
+	}
+
+	if err := os.Mkdir(filepath.Join(dir, "member"), 0o700); err != nil {
+		t.Fatal(err)
+	}
+	if etcdDataDirectoryIsFresh(dir) {
+		t.Fatal("any etcd state alongside the marker must block fresh bootstrap")
+	}
+}
+
+func TestEtcdDataDirectoryIsFreshRejectsMarkerDirectory(t *testing.T) {
+	dir := t.TempDir()
+	if err := os.Mkdir(filepath.Join(dir, ".etcd3_api"), 0o700); err != nil {
+		t.Fatal(err)
+	}
+	if etcdDataDirectoryIsFresh(dir) {
+		t.Fatal("only the regular marker file may be ignored")
+	}
+}
+
 func TestIdentityEndpointRequiresSharedProbeToken(t *testing.T) {
 	cfg := bootstrapTestConfig()
 
@@ -178,3 +206,43 @@ func TestIdentityEndpointRequiresSharedProbeToken(t *testing.T) {
 		t.Fatalf("identity = %+v, want app=%q node=%q", identity, cfg.AppName, cfg.MyName)
 	}
 }
+
+func TestIdentityEndpointReloadsCurrentCredentialsAndMembership(t *testing.T) {
+	startup := bootstrapTestConfig()
+	startup.PostgresReplicationPassword = "stale-secret"
+	startup.EtcdInitialCluster = "node-10-0-0-1=https://10.0.0.1:12380"
+
+	current := *startup
+	current.PostgresReplicationPassword = "current-secret"
+	current.EtcdInitialCluster = "node-10-0-0-1=https://10.0.0.1:12380," +
+		"node-10-0-0-2=https://10.0.0.2:12380," +
+		"node-10-0-0-3=https://10.0.0.3:12380"
+
+	recorder := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, clusterIdentityPath, nil)
+	req.Header.Set("X-Flux-Cluster-Probe", identityProbeToken(&current))
+	loads := 0
+	writeCurrentIdentityResponse(recorder, startup, req, func(cfg *config.Config) error {
+		loads++
+		*cfg = current
+		return nil
+	})
+
+	if loads != 1 {
+		t.Fatalf("cluster_env reloads = %d, want 1", loads)
+	}
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("identity status = %d, want %d", recorder.Code, http.StatusOK)
+	}
+	var identity nodeIdentity
+	if err := json.Unmarshal(recorder.Body.Bytes(), &identity); err != nil {
+		t.Fatal(err)
+	}
+	wantView := membershipNames(current.EtcdInitialCluster)
+	if !reflect.DeepEqual(identity.MembershipView, wantView) {
+		t.Fatalf("membership view = %v, want %v", identity.MembershipView, wantView)
+	}
+	if startup.PostgresReplicationPassword != "stale-secret" {
+		t.Fatal("identity reload mutated the proxy's shared startup config")
+	}
+}

cmd/flux-agent/etcd_start.go

@@ -334,19 +334,29 @@ func verifyLocalDataWithTempEtcd(cfg *config.Config, dataDir, token string, sslO
 // ETCD_INITIAL_CLUSTER string derived from actual registered members (needed
 // to satisfy etcd v3.3's "member count is unequal" check).
 func tryJoinExisting(cfg *config.Config, sslOpts, otherIPs []string, forceRejoin bool, maxRetries int) (string, string, error) {
-	delay := time.Duration(cfg.EtcdJoinRetryDelaySeconds) * time.Second
+	return tryJoinExistingWithLoader(cfg, sslOpts, otherIPs, forceRejoin, maxRetries, config.LoadClusterEnv)
+}
+
+func tryJoinExistingWithLoader(cfg *config.Config, sslOpts, otherIPs []string, forceRejoin bool, maxRetries int,
+	loadClusterEnv func(*config.Config) error) (string, string, error) {
 	if maxRetries < 1 {
 		maxRetries = 1
 	}
 
 	anyReachable := false
-	peerURL := fmt.Sprintf("%s://%s:%d", cfg.EtcdProtocol(), cfg.MyIP, cfg.HostEtcdPeerPort)
-
-	if len(otherIPs) == 0 {
-		return "", "", myerrors.NewNoPeersReachable()
-	}
 
 	for attempt := 1; attempt <= maxRetries; attempt++ {
+		// The reconciliation daemon may replace an incomplete startup topology
+		// while this process is waiting for Flux placement to settle. Refresh the
+		// inputs, not only the network operation, on every attempt. Keep the last
+		// usable snapshot if a concurrent read temporarily fails.
+		if err := loadClusterEnv(cfg); err != nil {
+			pkglog.Warnf("cannot refresh %s for peer discovery: %v", config.ClusterEnvFile, err)
+		} else {
+			otherIPs = otherIPsFromInitialCluster(cfg.EtcdInitialCluster, cfg.MyName)
+		}
+		delay := time.Duration(cfg.EtcdJoinRetryDelaySeconds) * time.Second
+		peerURL := fmt.Sprintf("%s://%s:%d", cfg.EtcdProtocol(), cfg.MyIP, cfg.HostEtcdPeerPort)
 		pkglog.Infof("peer discovery attempt %d/%d", attempt, maxRetries)
 		for _, ip := range otherIPs {
 			endpoint := fmt.Sprintf("%s://%s:%d", cfg.EtcdProtocol(), ip, cfg.HostEtcdClientPort)
@@ -451,7 +461,7 @@ func decideBootstrap(cfg *config.Config, dataDir string) string {
 	expectedCount := len(strings.Split(cfg.EtcdInitialCluster, ","))
 	etcdDataEmpty := !fileExists(dataDir + "/member/snap/db")
 	pgDataEmpty := !dirExists("/var/lib/postgresql/data/global")
-	strictlyEmpty := directoryIsEmpty(dataDir) && directoryIsEmpty("/var/lib/postgresql/data")
+	strictlyEmpty := etcdDataDirectoryIsFresh(dataDir) && directoryIsEmpty("/var/lib/postgresql/data")
 
 	candidate := bootstrapCandidate(cfg.EtcdInitialCluster)
 

cmd/flux-agent/proxy.go

@@ -257,7 +257,7 @@ func handleProxyConn(client net.Conn, primaryIP string, cfg *config.Config, host
 			return
 		}
 		defer req.Body.Close()
-		writeIdentityResponse(&connResponseWriter{Conn: client}, cfg, req)
+		writeCurrentIdentityResponse(&connResponseWriter{Conn: client}, cfg, req, config.LoadClusterEnv)
 		return
 	}
 	if primaryIP == "" {
@@ -284,6 +284,19 @@ func handleProxyConn(client net.Conn, primaryIP string, cfg *config.Config, host
 	<-done
 }
 
+// writeCurrentIdentityResponse reloads the configuration snapshot used for
+// authentication and membership reporting. The daemon can update cluster_env
+// without restarting the proxy, so serving the startup snapshot here can make
+// otherwise matching peers reject each other's probe tokens or topology.
+func writeCurrentIdentityResponse(w http.ResponseWriter, cfg *config.Config, r *http.Request,
+	loadClusterEnv func(*config.Config) error) {
+	current := *cfg
+	if err := loadClusterEnv(&current); err != nil {
+		pkglog.Warnf("proxy: could not refresh %s for identity probe: %v", config.ClusterEnvFile, err)
+	}
+	writeIdentityResponse(w, &current, r)
+}
+
 // connResponseWriter is the minimal http.ResponseWriter needed to serve the
 // identity document on the existing TCP proxy listener.
 type connResponseWriter struct {

cmd/flux-agent/recovery_test.go

@@ -2,13 +2,53 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/RunOnFlux/flux-pg-cluster/internal/config"
 )
 
+func TestJoinDiscoveryReloadsTopologyOnEveryAttempt(t *testing.T) {
+	cfg := recoveryTestConfig()
+	cfg.EtcdInitialCluster = "node-82-67-53-6=https://82.67.53.6:12380"
+	cfg.EtcdJoinRetryDelaySeconds = 0
+
+	loads := 0
+	_, _, err := tryJoinExistingWithLoader(cfg, nil, nil, false, 3, func(cfg *config.Config) error {
+		loads++
+		// Keep this as a self-only view so the test performs no network I/O. The
+		// important regression is that an initially empty peer list no longer
+		// returns before loading, and that all retry attempts refresh the input.
+		cfg.EtcdInitialCluster = "node-82-67-53-6=https://82.67.53.6:12380"
+		return nil
+	})
+	if err == nil {
+		t.Fatal("empty peer topology should still report no reachable peers")
+	}
+	if loads != 3 {
+		t.Fatalf("cluster_env reloads = %d, want one for each of 3 attempts", loads)
+	}
+}
+
+func TestJoinDiscoveryRetriesWhenReloadFailsWithNoTopology(t *testing.T) {
+	cfg := recoveryTestConfig()
+	cfg.EtcdJoinRetryDelaySeconds = 0
+
+	loads := 0
+	_, _, err := tryJoinExistingWithLoader(cfg, nil, nil, false, 2, func(*config.Config) error {
+		loads++
+		return errors.New("temporary read failure")
+	})
+	if err == nil {
+		t.Fatal("unreachable fallback peers should report no reachable peers")
+	}
+	if loads != 2 {
+		t.Fatalf("cluster_env reloads = %d, want 2", loads)
+	}
+}
+
 func TestBootstrapRetryDelayOutlivesSupervisorStartupWindow(t *testing.T) {
 	if got := bootstrapRetryDelay(0); got != 2*time.Second {
 		t.Fatalf("minimum retry delay = %s, want 2s", got)