From 7b3560c6421f1b40bff6fd164c705926db88c08c Mon Sep 17 00:00:00 2001 From: Philipp Winter Date: Mon, 31 Aug 2026 13:38:39 -0500 Subject: [PATCH 1/3] Fix race in primary election. The methods `Init`` and `PostInit`` each decided primary-vs-standby by reading the Consul "cluster initialized" flag and acting on it separately, with no atomicity between the two. When nodes booted close together, more than one could see the flag unset and independently self-initialize as primary, leaving a node with a fresh, credential-less local database that then failed forever trying to register as a standby against the real primary (repmgr/flypgadmin roles don't exist). Replace both checks with `Store.TryClaimPrimary``, a single atomic Consul CAS (ModifyIndex: 0) keyed by machine ID, so exactly one node ever wins the claim, and a node retrying after its own crash can tell it already won rather than deferring to itself as "someone else." --- internal/flypg/node.go | 26 +++++++++----------- internal/flypg/state/store.go | 46 ++++++++++++++++++++++++----------- 2 files changed, 44 insertions(+), 28 deletions(-) diff --git a/internal/flypg/node.go b/internal/flypg/node.go index fe252327..2de5fa38 100644 --- a/internal/flypg/node.go +++ b/internal/flypg/node.go @@ -139,8 +139,10 @@ func (n *Node) Init(ctx context.Context) error { return fmt.Errorf("failed initialize cluster state store: %s", err) } - // Check to see if cluster has already been initialized. - clusterInitialized, err := store.IsInitializationFlagSet() + // Atomically claim primary status for this cluster. Exactly one node + // across the whole app ever wins this; everyone else must join as a + // standby (or witness) against whoever did. + isPrimary, err := store.TryClaimPrimary(n.MachineID) if err != nil { return fmt.Errorf("failed to verify cluster state %s", err) } @@ -153,7 +155,7 @@ func (n *Node) Init(ctx context.Context) error { } // Remote restores are only eligible on uninitialized clusters. - if !clusterInitialized { + if isPrimary { // Determine if we are performing a remote restore. if err := n.handleRemoteRestore(ctx, store); err != nil { return fmt.Errorf("failed to handle remote restore: %s", err) @@ -176,7 +178,7 @@ func (n *Node) Init(ctx context.Context) error { } if !n.PGConfig.isInitialized() { - if clusterInitialized { + if !isPrimary { if n.RepMgr.Witness { log.Println("Provisioning witness") if err := n.PGConfig.writePasswordFile(n.OperatorCredentials.Password); err != nil { @@ -346,19 +348,21 @@ func (n *Node) PostInit(ctx context.Context) error { } else { // New member - // Check with consul to see if the cluster has already been initialized + // Check with consul to see if this node is the cluster's primary. This + // is the same durable claim made in Init - calling it again here is + // idempotent (same machineID, same stored owner) and correctly + // resumes as primary across a crash/restart between Init and here. store, err := state.NewStore() if err != nil { return fmt.Errorf("failed initialize cluster state store. %v", err) } - // The initialization flag is set after the primary is registered. - clusterInitialized, err := store.IsInitializationFlagSet() + isPrimary, err := store.TryClaimPrimary(n.MachineID) if err != nil { return fmt.Errorf("failed to verify cluster state: %s", err) } - if !clusterInitialized { + if isPrimary { // Configure as primary log.Println("Registering primary") @@ -385,12 +389,6 @@ func (n *Node) PostInit(ctx context.Context) error { return fmt.Errorf("failed to register repmgr primary: %s", err) } - // Set initialization flag within consul so future members know they are joining - // an existing cluster. - if err := store.SetInitializationFlag(); err != nil { - return fmt.Errorf("failed to register cluster with consul") - } - // Let the boot process know that we've already been configured. if err := issueRegistrationCert(); err != nil { return fmt.Errorf("failed to issue registration certificate: %s", err) diff --git a/internal/flypg/state/store.go b/internal/flypg/state/store.go index 2efe9458..54ec46df 100644 --- a/internal/flypg/state/store.go +++ b/internal/flypg/state/store.go @@ -36,23 +36,41 @@ func NewStore() (*Store, error) { }, nil } -func (c *Store) SetInitializationFlag() error { - kv := &api.KVPair{Key: c.targetKey("INITIALIZED"), Value: []byte("true")} - _, err := c.Client.KV().Put(kv, nil) - return err -} - -func (c *Store) IsInitializationFlagSet() (bool, error) { - result, _, err := c.Client.KV().Get(c.targetKey("INITIALIZED"), nil) +// TryClaimPrimary attempts to atomically claim this cluster for the given +// machine. It replaces the old check-then-act pattern (read the flag, then +// separately decide to become primary), which raced whenever more than one +// node's Init/PostInit ran the check before the eventual primary had gotten +// around to setting the flag - both would see it unset and both would try +// to become primary. +// +// Consul's CAS with ModifyIndex: 0 only succeeds if the key does not already +// exist, so when multiple nodes call this concurrently, exactly one write +// wins across the whole cluster - Consul's Raft log serializes it, so there +// is no window between "check" and "act" for another node to slip through. +// +// Returns true if this machine is the primary: either it just won the +// claim, or it already won an earlier attempt (e.g. retrying after its own +// crash), recognized by the stored value matching its own machineID rather +// than someone else's. +func (c *Store) TryClaimPrimary(machineID string) (bool, error) { + key := c.targetKey("INITIALIZED") + + ok, _, err := c.Client.KV().CAS(&api.KVPair{ + Key: key, + Value: []byte(machineID), + ModifyIndex: 0, + }, nil) if err != nil { - return false, err + return false, fmt.Errorf("failed to claim primary: %s", err) } - - if result == nil { - return false, nil + if ok { + return true, nil } - - return true, nil + pair, _, err := c.Client.KV().Get(key, nil) + if err != nil { + return false, fmt.Errorf("failed to verify primary claim: %s", err) + } + return pair != nil && string(pair.Value) == machineID, nil } func (c *Store) PushUserConfig(key string, config []byte) error { From 0e4e4d13bec707e161b6efba831220d7c23f9f55 Mon Sep 17 00:00:00 2001 From: Philipp Winter Date: Mon, 31 Aug 2026 14:08:57 -0500 Subject: [PATCH 2/3] Pacify the linter. --- internal/flypg/state/store.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/flypg/state/store.go b/internal/flypg/state/store.go index 54ec46df..e56cd16d 100644 --- a/internal/flypg/state/store.go +++ b/internal/flypg/state/store.go @@ -70,6 +70,7 @@ func (c *Store) TryClaimPrimary(machineID string) (bool, error) { if err != nil { return false, fmt.Errorf("failed to verify primary claim: %s", err) } + return pair != nil && string(pair.Value) == machineID, nil } From f028570cab2692b4680c3b0ccc3f8f018f301418 Mon Sep 17 00:00:00 2001 From: Philipp Winter Date: Mon, 31 Aug 2026 15:03:00 -0500 Subject: [PATCH 3/3] Be patient while waiting for primary to be ready. --- internal/flypg/node.go | 24 +++++++++++++----------- internal/flypg/repmgr.go | 29 +++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 11 deletions(-) diff --git a/internal/flypg/node.go b/internal/flypg/node.go index 2de5fa38..d5744423 100644 --- a/internal/flypg/node.go +++ b/internal/flypg/node.go @@ -142,7 +142,7 @@ func (n *Node) Init(ctx context.Context) error { // Atomically claim primary status for this cluster. Exactly one node // across the whole app ever wins this; everyone else must join as a // standby (or witness) against whoever did. - isPrimary, err := store.TryClaimPrimary(n.MachineID) + isPrimary, err := n.tryClaimPrimary(store) if err != nil { return fmt.Errorf("failed to verify cluster state %s", err) } @@ -190,7 +190,7 @@ func (n *Node) Init(ctx context.Context) error { } } else { log.Println("Provisioning standby") - cloneTarget, err := n.RepMgr.ResolvePrimaryOverDNS(ctx) + cloneTarget, err := n.RepMgr.WaitForPrimaryOverDNS(ctx) if err != nil { return fmt.Errorf("failed to resolve member over dns: %s", err) } @@ -357,7 +357,7 @@ func (n *Node) PostInit(ctx context.Context) error { return fmt.Errorf("failed initialize cluster state store. %v", err) } - isPrimary, err := store.TryClaimPrimary(n.MachineID) + isPrimary, err := n.tryClaimPrimary(store) if err != nil { return fmt.Errorf("failed to verify cluster state: %s", err) } @@ -366,14 +366,6 @@ func (n *Node) PostInit(ctx context.Context) error { // Configure as primary log.Println("Registering primary") - // Verify we reside within the clusters primary region - if !n.RepMgr.eligiblePrimary() { - return fmt.Errorf("unable to configure as the primary. expected region: %q, got: %q", - n.PrimaryRegion, - n.RepMgr.Region, - ) - } - // Create required users if err := n.setupCredentials(ctx, conn); err != nil { return fmt.Errorf("failed to create required users: %s", err) @@ -437,6 +429,16 @@ func (n *Node) PostInit(ctx context.Context) error { return nil } +// tryClaimPrimary ensures only non-witness nodes in the primary region can +// participate in the initial primary election. +func (n *Node) tryClaimPrimary(store *state.Store) (bool, error) { + if n.RepMgr.Witness || !n.RepMgr.eligiblePrimary() { + return false, nil + } + + return store.TryClaimPrimary(n.MachineID) +} + func (n *Node) setupCredentials(ctx context.Context, conn *pgx.Conn) error { requiredCredentials := []admin.Credential{ n.OperatorCredentials, diff --git a/internal/flypg/repmgr.go b/internal/flypg/repmgr.go index 4714d275..74a99bb3 100644 --- a/internal/flypg/repmgr.go +++ b/internal/flypg/repmgr.go @@ -11,6 +11,7 @@ import ( "os" "strconv" "strings" + "time" "github.com/fly-apps/postgres-flex/internal/privnet" "github.com/fly-apps/postgres-flex/internal/utils" @@ -26,6 +27,9 @@ const ( UnknownRoleName = "" repmgrConsulKey = "repmgr" + + primaryResolutionRetryInterval = time.Second + primaryResolutionTimeout = time.Minute ) type RepMgr struct { @@ -512,6 +516,31 @@ func (r *RepMgr) ResolvePrimaryOverDNS(ctx context.Context) (*Member, error) { return target, nil } +// WaitForPrimaryOverDNS waits for the elected primary to finish initializing +// and become available for cloning. +func (r *RepMgr) WaitForPrimaryOverDNS(ctx context.Context) (*Member, error) { + waitCtx, cancel := context.WithTimeout(ctx, primaryResolutionTimeout) + defer cancel() + + ticker := time.NewTicker(primaryResolutionRetryInterval) + defer ticker.Stop() + + var lastErr error + for { + primary, err := r.ResolvePrimaryOverDNS(waitCtx) + if err == nil { + return primary, nil + } + lastErr = err + + select { + case <-waitCtx.Done(): + return nil, fmt.Errorf("waiting for cloneable primary: %w (last error: %v)", waitCtx.Err(), lastErr) + case <-ticker.C: + } + } +} + func (r *RepMgr) InRegionPeerIPs(ctx context.Context) ([]net.IPAddr, error) { targets := fmt.Sprintf("%s.%s", r.PrimaryRegion, r.AppName) return privnet.AllPeers(ctx, targets)