Recover cluster leader from Postgres failover

pkg/rselection/impls/pgx/leader.go

@@ -53,13 +53,27 @@ type PgxLeader struct {
 	pingSuccess     bool
 	pingSuccessLock sync.RWMutex
 
+	// stepDownTimeout is the duration of continuous cluster-integrity failure
+	// after which the leader steps down so a fresh election can occur. Zero
+	// disables step-down (legacy behavior).
+	stepDownTimeout time.Duration
+
+	// unhealthySince is the time the leader most recently entered an unhealthy
+	// state, or the zero value when currently healthy. Only accessed from the
+	// lead() goroutine.
+	unhealthySince time.Time
+
+	// now returns the current time; overridable in tests. Use timeNow().
+	now func() time.Time
+
 	// Stores information about active nodes and ping times
 	nodes map[string]*electiontypes.ClusterNode
 	mutex sync.RWMutex
 
 	// Used for testing
 	loopAwareChTEST    chan bool
 	pingResponseChTEST chan bool
+	setClockTEST       func(time.Time)
 }
 
 type PgxLeaderConfig struct {
@@ -75,23 +89,26 @@ type PgxLeaderConfig struct {
 	PingInterval    time.Duration
 	SweepInterval   time.Duration
 	MaxPingAge      time.Duration
+	StepDownTimeout time.Duration
 }
 
 func NewPgxLeader(cfg PgxLeaderConfig) *PgxLeader {
 	return &PgxLeader{
-		store:       cfg.Store,
-		awb:         cfg.Broadcaster,
-		notify:      cfg.Notifier,
-		taskHandler: cfg.TaskHandler,
-		chLeader:    cfg.LeaderChannel,
-		chFollower:  cfg.FollowerChannel,
-		chMessages:  cfg.MessagesChannel,
-		address:     cfg.Address,
-		stop:        cfg.StopChan,
-		ping:        cfg.PingInterval,
-		sweep:       cfg.SweepInterval,
-		maxPingAge:  cfg.MaxPingAge,
-		mutex:       sync.RWMutex{},
+		store:           cfg.Store,
+		awb:             cfg.Broadcaster,
+		notify:          cfg.Notifier,
+		taskHandler:     cfg.TaskHandler,
+		chLeader:        cfg.LeaderChannel,
+		chFollower:      cfg.FollowerChannel,
+		chMessages:      cfg.MessagesChannel,
+		address:         cfg.Address,
+		stop:            cfg.StopChan,
+		ping:            cfg.PingInterval,
+		sweep:           cfg.SweepInterval,
+		maxPingAge:      cfg.MaxPingAge,
+		stepDownTimeout: cfg.StepDownTimeout,
+		now:             time.Now,
+		mutex:           sync.RWMutex{},
 	}
 }
 
@@ -175,6 +192,10 @@ func (p *PgxLeader) lead(ctx context.Context, pingTick, sweepTick <-chan time.Ti
 			go p.pingNodes(ctx)
 		case <-sweepTick:
 			p.sweepNodes()
+			if p.evaluateHealth() {
+				slog.Error("Leader stepping down after sustained cluster integrity failure; a new election will occur", "timeout", p.stepDownTimeout)
+				return
+			}
 		case vCh := <-p.taskHandler.Verify():
 			p.verify(vCh)
 		case <-logTickCh:
@@ -201,58 +222,99 @@ func (p *PgxLeader) info() string {
 	return fmt.Sprintf("Cluster nodes:\n  %s", strings.Join(nodes, "\n  "))
 }
 
-// verify ensures that the in-memory node list matches the store list. We do
-// this to ensure that the cluster is healthy before running scheduled tasks.
-// This helps to prevent split-brain issues in the cluster.
-//
-//  1. The task handler is ready to run a scheduled task.
-//  2. The task handler sends a `chan bool` to its verify channel.
-//  3. We receive the `chan bool` here and:
-//     a. Verify that the cluster is healthy.
-//     b. Respond with `true` over the channel if the cluster is healthy.
-//  4. The task handler runs the scheduled task when the cluster is healthy.
-func (p *PgxLeader) verify(vCh chan bool) {
-	var err error
-
-	// Upon exit, notify channel with `true` or `false` depending upon error status
-	defer func(err *error) {
-		if *err != nil {
-			// TODO this should be worded better because sometimes this error is normal and expected,
-			// e.g. when restarting a cluster node and the node list is temporarily out of sync:
-			// Error verifying cluster integrity: node list length differs. Store node count 3 does not match leader count 2
-			// It would be great to be able to distinguish between expected/unexpected errors here and log accordingly.
-			slog.Error("Error verifying cluster integrity", "error", *err)
-			vCh <- false
-		} else {
-			vCh <- true
-		}
-	}(&err)
-
+// clusterIntegrityErr reports whether the leader is in contact with every node
+// the store considers part of the cluster, returning nil when healthy. As a
+// side effect it reconciles the in-memory node map asymmetrically: nodes that
+// have left the store are pruned (always safe), but store-only nodes are NOT
+// added here. Reachable nodes are re-added by handlePingResponse when they
+// answer a ping, which preserves the split-brain guarantee — the leader only
+// reports healthy once it is actually in contact with every registered node.
+func (p *PgxLeader) clusterIntegrityErr() error {
 	if p.unsuccessfulPing() {
-		err = fmt.Errorf("error pinging follower nodes")
-		return
+		return fmt.Errorf("error pinging follower nodes")
 	}
 
-	// Enumerate nodes recorded in database
 	nodes, err := p.store.Nodes()
 	if err != nil {
-		err = fmt.Errorf("error retrieving cluster node list for verification: %s", err)
-		return
+		return fmt.Errorf("error retrieving cluster node list for verification: %w", err)
 	}
 
-	// Ensure in-memory node list length matches store
-	if len(nodes) != len(p.nodes) {
-		err = fmt.Errorf("node list length differs. Store node count %d does not match leader count %d", len(nodes), len(p.nodes))
-		return
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
+	// Build the set of node keys the store considers part of the cluster.
+	storeKeys := make(map[string]struct{}, len(nodes))
+	for _, node := range nodes {
+		storeKeys[node.Key()] = struct{}{}
+	}
+
+	// Prune in-memory nodes that are no longer present in the store.
+	// Unlike sweepNodes, we do not exempt the leader's own key here: the store
+	// includes the leader, and even if a transient store read omitted it the
+	// leader re-adds itself via its self-ping, so pruning it is self-healing.
+	for key := range p.nodes {
+		if _, ok := storeKeys[key]; !ok {
+			slog.Info("Leader pruning node no longer present in store", "node", key)
+			delete(p.nodes, key)
+		}
 	}
 
-	// Ensure in-memory node list matches store list
+	// Any store node still missing from memory means the leader is not yet in
+	// contact with a registered node.
 	for _, node := range nodes {
 		if _, ok := p.nodes[node.Key()]; !ok {
-			err = fmt.Errorf("node %s with IP %s from store not known by leader", node.Name, node.IP)
-			return
+			return fmt.Errorf("node %s with IP %s from store not known by leader", node.Name, node.IP)
 		}
 	}
+
+	return nil
+}
+
+// evaluateHealth runs the cluster integrity check and tracks how long the
+// leader has been unhealthy. It returns true when the leader should step down:
+// the cluster has been continuously unhealthy for at least stepDownTimeout
+// (when that timeout is non-zero). Called from the lead() goroutine only.
+func (p *PgxLeader) evaluateHealth() bool {
+	err := p.clusterIntegrityErr()
+	if err == nil {
+		if !p.unhealthySince.IsZero() {
+			slog.Info("Cluster integrity recovered", "degradedFor", p.timeNow().Sub(p.unhealthySince))
+		}
+		p.unhealthySince = time.Time{}
+		return false
+	}
+
+	// Log the onset of an unhealthy episode once, at INFO, so it appears in
+	// default logs (and is available for a customer escalation) without raising
+	// the log level. Per-tick detail stays at debug to avoid noise on routine,
+	// self-healing mismatches such as a rolling restart. A sustained failure
+	// escalates to error, but only when step-down is enabled: a disabled leader
+	// takes no recovery action and would otherwise log error every tick forever.
+	if p.unhealthySince.IsZero() {
+		p.unhealthySince = p.timeNow()
+		slog.Info("Cluster integrity check failing", "error", err)
+	}
+	dur := p.timeNow().Sub(p.unhealthySince)
+
+	if p.stepDownTimeout > 0 && dur >= p.stepDownTimeout/2 {
+		slog.Error("Cluster integrity check still failing", "error", err, "unhealthyFor", dur)
+	} else {
+		slog.Debug("Cluster integrity check failing", "error", err, "unhealthyFor", dur)
+	}
+
+	return p.stepDownTimeout > 0 && dur >= p.stepDownTimeout
+}
+
+// verify gates a scheduled task on cluster health. It responds true on the
+// channel when the cluster is healthy, false otherwise. Leveled alerting on
+// sustained failure lives in evaluateHealth, so this logs only at debug.
+func (p *PgxLeader) verify(vCh chan bool) {
+	if err := p.clusterIntegrityErr(); err != nil {
+		slog.Debug("Skipping scheduled task; cluster integrity check failed", "error", err)
+		vCh <- false
+		return
+	}
+	vCh <- true
 }
 
 // pingNodes sends a ping request out on the follower channel. All online cluster
@@ -392,3 +454,13 @@ func (p *PgxLeader) unsuccessfulPing() bool {
 	defer p.pingSuccessLock.RUnlock()
 	return !p.pingSuccess
 }
+
+// timeNow returns the current time, using the injectable p.now when set. This
+// keeps PgxLeader instances constructed as struct literals (e.g., in tests)
+// safe to use without setting now.
+func (p *PgxLeader) timeNow() time.Time {
+	if p.now != nil {
+		return p.now()
+	}
+	return time.Now()
+}

pkg/rselection/impls/pgx/leader_recovery_test.go

@@ -0,0 +1,143 @@
+package pgxelection
+
+// Copyright (C) 2026 by Posit Software, PBC
+
+import (
+	"errors"
+	"time"
+
+	"gopkg.in/check.v1"
+
+	"github.com/rstudio/platform-lib/v4/pkg/rselection/electiontypes"
+)
+
+// RecoverySuite holds Postgres-free unit tests for the leader's integrity
+// check and step-down logic. It intentionally does not use the shared pgx
+// pool, so it runs even under `go test -short`.
+type RecoverySuite struct{}
+
+var _ = check.Suite(&RecoverySuite{})
+
+func newTestLeader(store ClusterPgStore, nodes map[string]*electiontypes.ClusterNode) *PgxLeader {
+	return &PgxLeader{
+		store:       store,
+		notify:      FakePgNotifier{},
+		nodes:       nodes,
+		pingSuccess: true,
+	}
+}
+
+// A node present in memory but absent from the store is pruned, and the check
+// then reports healthy.
+func (s *RecoverySuite) TestIntegrityPrunesDepartedNode(c *check.C) {
+	store := &fakeStore{nodes: map[string]*electiontypes.ClusterNode{
+		"a": {Name: "a", IP: "10.0.0.1"},
+	}}
+	leader := newTestLeader(store, map[string]*electiontypes.ClusterNode{
+		"a_10.0.0.1": {Name: "a", IP: "10.0.0.1"},
+		"b_10.0.0.2": {Name: "b", IP: "10.0.0.2"}, // not in store; should be pruned
+	})
+
+	c.Assert(leader.clusterIntegrityErr(), check.IsNil)
+	c.Assert(leader.nodes, check.HasLen, 1)
+	_, ok := leader.nodes["b_10.0.0.2"]
+	c.Assert(ok, check.Equals, false)
+}
+
+// A node present in the store but absent from memory is an error and is NOT
+// added to memory (it must be re-added by a real ping response).
+func (s *RecoverySuite) TestIntegrityStoreOnlyNodeErrorsWithoutAdding(c *check.C) {
+	store := &fakeStore{nodes: map[string]*electiontypes.ClusterNode{
+		"a": {Name: "a", IP: "10.0.0.1"},
+		"b": {Name: "b", IP: "10.0.0.2"},
+	}}
+	leader := newTestLeader(store, map[string]*electiontypes.ClusterNode{
+		"a_10.0.0.1": {Name: "a", IP: "10.0.0.1"},
+	})
+
+	c.Assert(leader.clusterIntegrityErr(), check.NotNil)
+	_, ok := leader.nodes["b_10.0.0.2"]
+	c.Assert(ok, check.Equals, false)
+}
+
+func (s *RecoverySuite) TestIntegrityUnsuccessfulPingErrors(c *check.C) {
+	leader := newTestLeader(&fakeStore{}, map[string]*electiontypes.ClusterNode{})
+	leader.pingSuccess = false
+	c.Assert(leader.clusterIntegrityErr(), check.NotNil)
+}
+
+func (s *RecoverySuite) TestIntegrityStoreErrorErrors(c *check.C) {
+	store := &fakeStore{err: errors.New("db down")}
+	leader := newTestLeader(store, map[string]*electiontypes.ClusterNode{})
+	c.Assert(leader.clusterIntegrityErr(), check.NotNil)
+}
+
+// healthyLeader returns a leader whose integrity check passes (ping ok, store
+// matches memory) with an injectable clock starting at t.
+func healthyLeader(t time.Time, stepDown time.Duration) *PgxLeader {
+	store := &fakeStore{nodes: map[string]*electiontypes.ClusterNode{
+		"a": {Name: "a", IP: "10.0.0.1"},
+	}}
+	leader := newTestLeader(store, map[string]*electiontypes.ClusterNode{
+		"a_10.0.0.1": {Name: "a", IP: "10.0.0.1"},
+	})
+	leader.stepDownTimeout = stepDown
+	cur := t
+	leader.now = func() time.Time { return cur }
+	leader.setClockTEST = func(nt time.Time) { cur = nt }
+	return leader
+}
+
+func (s *RecoverySuite) TestStepDownAfterSustainedFailure(c *check.C) {
+	start := time.Unix(1_000_000, 0)
+	leader := healthyLeader(start, 30*time.Second)
+
+	// Healthy: no step-down, unhealthySince stays zero.
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+	c.Assert(leader.unhealthySince.IsZero(), check.Equals, true)
+
+	// Force unhealthy via failing pings.
+	leader.pingSuccess = false
+
+	// First failure records the time but does not step down.
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+	c.Assert(leader.unhealthySince.Equal(start), check.Equals, true)
+
+	// Still within the timeout window.
+	leader.setClockTEST(start.Add(29 * time.Second))
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+
+	// At/after the timeout, step down.
+	leader.setClockTEST(start.Add(30 * time.Second))
+	c.Assert(leader.evaluateHealth(), check.Equals, true)
+}
+
+func (s *RecoverySuite) TestStepDownResetsOnRecovery(c *check.C) {
+	start := time.Unix(1_000_000, 0)
+	leader := healthyLeader(start, 30*time.Second)
+
+	leader.pingSuccess = false
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+	c.Assert(leader.unhealthySince.IsZero(), check.Equals, false)
+
+	// Recover before the timeout elapses.
+	leader.pingSuccess = true
+	leader.setClockTEST(start.Add(10 * time.Second))
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+	c.Assert(leader.unhealthySince.IsZero(), check.Equals, true)
+
+	// A later, separate failure must not immediately trip the timeout.
+	leader.pingSuccess = false
+	leader.setClockTEST(start.Add(11 * time.Second))
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+}
+
+func (s *RecoverySuite) TestStepDownDisabledByZeroTimeout(c *check.C) {
+	start := time.Unix(1_000_000, 0)
+	leader := healthyLeader(start, 0) // disabled
+	leader.pingSuccess = false
+
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+	leader.setClockTEST(start.Add(100 * time.Hour))
+	c.Assert(leader.evaluateHealth(), check.Equals, false)
+}