kvserver: add leaderlessWatcher

This commit adds a leaderlessWatcher to the replica. This will be used
to signal when that replica has been leaderless for a long time.

This signal will be used in future commits to return a
ReplicaUnavailable error to requests that encounter replicas without a
leader for a long time. This indicates that the range have lost quorum,
and can't establish leadership in time.

This is important for leader leases in particular because replicas will
not propose a lease if they are not the leader, which makes the replica
circuit breakers not effective when it comes to quorum loss with leader
leases.

References: #139638

Release note: None

Author: iskettaneh

docs/generated/settings/settings.html

@@ -116,6 +116,7 @@
 <tr><td><div id="setting-kv-rangefeed-closed-timestamp-refresh-interval" class="anchored"><code>kv.rangefeed.closed_timestamp_refresh_interval</code></div></td><td>duration</td><td><code>3s</code></td><td>the interval at which closed-timestamp updatesare delivered to rangefeeds; set to 0 to use kv.closed_timestamp.side_transport_interval</td><td>Dedicated/Self-hosted (read-write); Serverless (read-only)</td></tr>
 <tr><td><div id="setting-kv-rangefeed-enabled" class="anchored"><code>kv.rangefeed.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>if set, rangefeed registration is enabled</td><td>Dedicated/Self-hosted (read-write); Serverless (read-only)</td></tr>
 <tr><td><div id="setting-kv-replica-circuit-breaker-slow-replication-threshold" class="anchored"><code>kv.replica_circuit_breaker.slow_replication_threshold</code></div></td><td>duration</td><td><code>1m0s</code></td><td>duration after which slow proposals trip the per-Replica circuit breaker (zero duration disables breakers)</td><td>Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-kv-replica-raft-leaderless-unavailable-threshold" class="anchored"><code>kv.replica_raft.leaderless_unavailable_threshold</code></div></td><td>duration</td><td><code>1m0s</code></td><td>duration after which leaderless replicas is considered unavailable. Set to 0 to disable leaderless replica availability checks</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-replica-stats-addsst-request-size-factor" class="anchored"><code>kv.replica_stats.addsst_request_size_factor</code></div></td><td>integer</td><td><code>50000</code></td><td>the divisor that is applied to addsstable request sizes, then recorded in a leaseholders QPS; 0 means all requests are treated as cost 1</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-replication-reports-interval" class="anchored"><code>kv.replication_reports.interval</code></div></td><td>duration</td><td><code>1m0s</code></td><td>the frequency for generating the replication_constraint_stats, replication_stats_report and replication_critical_localities reports (set to 0 to disable)</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-snapshot-rebalance-max-rate" class="anchored"><code>kv.snapshot_rebalance.max_rate</code></div></td><td>byte size</td><td><code>32 MiB</code></td><td>the rate limit (bytes/sec) to use for rebalance and upreplication snapshots</td><td>Dedicated/Self-Hosted</td></tr>

pkg/kv/kvserver/replica.go

@@ -176,6 +176,62 @@ func (c *atomicConnectionClass) set(cc rpc.ConnectionClass) {
 	atomic.StoreUint32((*uint32)(c), uint32(cc))
 }
 
+// leaderlessWatcher is a lightweight implementation of the signaller interface
+// that is used to signal when a replica doesn't know who the leader is for an
+// extended period of time. This is used to signal that the range in
+// unavailable.
+type leaderlessWatcher struct {
+	mu struct {
+		syncutil.RWMutex
+
+		// leaderlessTimestamp records the timestamp captured when the replica
+		// didn't know who the leader was. This is reset on every tick if the
+		// replica knows who the leader is.
+		leaderlessTimestamp time.Time
+
+		// unavailable is set to true if the replica is leaderless for a long time
+		// (longer than ReplicaLeaderlessUnavailableThreshold).
+		unavailable bool
+	}
+
+	// err is the error returned when the replica is leaderless for a long time.
+	err error
+
+	// closedChannel is an already closed channel. Requests will use it to know
+	// that the replica is leaderless, and can be considered unavailable. This
+	// is primarily due to implementation details of the request path, where
+	// the request grabs a signaller in signallerForBatch() and then checks if
+	// the channel is closed to determine if the replica is available.
+	closedChannel chan struct{}
+}
+
+// NewLeaderlessWatcher initializes a new leaderlessWatcher with the default
+// values.
+func NewLeaderlessWatcher(r *Replica) *leaderlessWatcher {
+	closedCh := make(chan struct{})
+	close(closedCh)
+	return &leaderlessWatcher{
+		err: r.replicaUnavailableError(
+			errors.Errorf("replica has been leaderless for %s",
+				ReplicaLeaderlessUnavailableThreshold.Get(&r.store.cfg.Settings.SV))),
+		closedChannel: closedCh,
+	}
+}
+
+func (lw *leaderlessWatcher) Err() error {
+	return lw.err
+}
+
+func (lw *leaderlessWatcher) C() <-chan struct{} {
+	return lw.closedChannel
+}
+
+func (lw *leaderlessWatcher) IsUnavailable() bool {
+	lw.mu.RLock()
+	defer lw.mu.RUnlock()
+	return lw.mu.unavailable
+}
+
 // ReplicaMutex is an RWMutex. It has its own type to make it easier to look for
 // usages specific to the replica mutex.
 type ReplicaMutex syncutil.RWMutex
@@ -941,6 +997,10 @@ type Replica struct {
 		lastTickTimestamp hlc.ClockTimestamp
 	}
 
+	// LeaderlessWatcher is used to signal when a replica is leaderless for a long
+	// time.
+	LeaderlessWatcher *leaderlessWatcher
+
 	// The raft log truncations that are pending. Access is protected by its own
 	// mutex. All implementation details should be considered hidden except to
 	// the code in raft_log_truncator.go. External code should only use the

pkg/kv/kvserver/replica_init.go

@@ -245,6 +245,7 @@ func newUninitializedReplicaWithoutRaftGroup(
 	r.breaker = newReplicaCircuitBreaker(
 		store.cfg.Settings, store.stopper, r.AmbientContext, r, onTrip, onReset,
 	)
+	r.LeaderlessWatcher = NewLeaderlessWatcher(r)
 	r.mu.currentRACv2Mode = r.replicationAdmissionControlModeToUse(context.TODO())
 	r.raftMu.flowControlLevel = kvflowcontrol.GetV2EnabledWhenLeaderLevel(
 		r.raftCtx, store.ClusterSettings(), store.TestingKnobs().FlowControlTestingKnobs)

pkg/kv/kvserver/replica_raft.go

@@ -32,6 +32,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/raft/tracker"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util"
@@ -54,6 +55,20 @@ import (
 var raftDisableLeaderFollowsLeaseholder = envutil.EnvOrDefaultBool(
 	"COCKROACH_DISABLE_LEADER_FOLLOWS_LEASEHOLDER", false)
 
+// ReplicaLeaderlessUnavailableThreshold is the duration after which leaderless
+// replicas are considered unavailable. Set to 0 to disable.
+var ReplicaLeaderlessUnavailableThreshold = settings.RegisterDurationSettingWithExplicitUnit(
+	settings.SystemOnly,
+	"kv.replica_raft.leaderless_unavailable_threshold",
+	"duration after which leaderless replicas is considered unavailable. Set to 0"+
+		" to disable leaderless replica availability checks",
+	60*time.Second,
+	settings.WithPublic,
+	// Setting the duration too low could be very dangerous to cluster health as
+	// replicas under normal operation could be considered unavailable.
+	settings.DurationWithMinimumOrZeroDisable(5*time.Second),
+)
+
 // evalAndPropose prepares the necessary pending command struct and initializes
 // a client command ID if one hasn't been. A verified lease is supplied as a
 // parameter if the command requires a lease; nil otherwise. It then evaluates

pkg/kv/kvserver/replica_test.go

@@ -15400,3 +15400,43 @@ func TestLockAcquisitions1PCInteractions(t *testing.T) {
 		})
 	})
 }
+
+// TestLeaderlessWatcherInit tests that the leaderless watcher is initialized
+// correctly.
+func TestLeaderlessWatcherInit(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	tc := testContext{}
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+
+	// Set the leaderless threshold to 10 second.
+	tsc := TestStoreConfig(nil /* clock */)
+	ReplicaLeaderlessUnavailableThreshold.Override(ctx, &tsc.Settings.SV, 10*time.Second)
+	tc.StartWithStoreConfig(ctx, t, stopper, tsc)
+
+	repl, err := tc.store.GetReplica(1)
+	require.NoError(t, err)
+
+	repl.LeaderlessWatcher.mu.RLock()
+	defer repl.LeaderlessWatcher.mu.RUnlock()
+
+	// Initially, the leaderWatcher doesn't consider the replica as unavailable.
+	require.False(t, repl.LeaderlessWatcher.IsUnavailable())
+
+	// The leaderless timestamp is not set.
+	require.Equal(t, time.Time{}, repl.LeaderlessWatcher.mu.leaderlessTimestamp)
+
+	// The error is always loaded.
+	require.Regexp(t, "replica has been leaderless for 10s", repl.LeaderlessWatcher.Err())
+
+	// The channel is closed.
+	c := repl.LeaderlessWatcher.C()
+	select {
+	case <-c:
+		// Channel is closed, which is expected
+	default:
+		t.Fatalf("expected LeaderlessWatcher channel to be closed")
+	}
+}