kvserver: mark replica as unavailable if leaderless for a long time

This commit marks the replica as unavailable in the leaderlessWatcher if
it has been leaderless for a duration above:
`kv.replica_raft.leaderless_unavailable_threshold`. This helps requests
to bail early on unavailable ranges without relying on the replica
circuit breaker to trip. This has multiple benefits:

1) Faster reaction time than the replica circuit breaker: If two nodes
fail (assuming R=3), many ranges will become unavailable. With the
replica circuit breaker, the breaker is tripped on the request path,
causing added delays. However, with this approach, the replica will
basically become unavailable on the tick path (rather than on the
request path).

2) Lighter weight: Instead of relying on the replica circuit breaker
to test the replication pipeline before it marks the range as
available again, this approach relies on the Raft signal to know
when there is a leader again, indicating that the range is
available again.

3) With leader leases, a replica won't propose a lease if it's not
the leader. This means that with leader leases, the replica circuit
breaker might not trip if the range have lost quorum. However, with
this commit, the replica will eventually forget who the leader was,
and eventually the leaderlessWatcher would mark it as unavailable.

Fixes: #139638

Release note: None

Author: iskettaneh

pkg/kv/kvserver/client_replica_test.go

@@ -39,6 +39,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts/ptutil"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/raftutil"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/stateloader"
+	"github.com/cockroachdb/cockroach/pkg/raft"
 	"github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/server"
@@ -2728,6 +2729,142 @@ func TestChangeReplicasGeneration(t *testing.T) {
 	assert.EqualValues(t, repl.Desc().Generation, oldGeneration+3, "\nold: %+v\nnew: %+v", oldDesc, newDesc)
 }
 
+// TestLossQuorumCauseLeaderlessWatcherToSignalUnavailable checks that if a range
+// lost its quorum, the remaining replicas in that range will have their
+// leaderlessWatcher indicate that the range is unavailable. Also, it checks
+// that when the range regains quorum, the leaderlessWatcher will indicate that
+// the range is available.
+func TestLossQuorumCauseLeaderWatcherToSignalUnavailable(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	manualClock := hlc.NewHybridManualClock()
+
+	stickyVFSRegistry := fs.NewStickyRegistry()
+	lisReg := listenerutil.NewListenerRegistry()
+	defer lisReg.Close()
+
+	// Perform a basic test setup with two nodes.
+	const numServers int = 2
+	st := cluster.MakeTestingClusterSettings()
+
+	// Set `kv.replica_raft.leaderless_unavailable_threshold` to 10 seconds.
+	threshold := time.Second * 10
+	kvserver.ReplicaLeaderlessUnavailableThreshold.Override(ctx, &st.SV, threshold)
+
+	stickyServerArgs := make(map[int]base.TestServerArgs)
+	for i := 0; i < numServers; i++ {
+		stickyServerArgs[i] = base.TestServerArgs{
+			Settings: st,
+			StoreSpecs: []base.StoreSpec{
+				{
+					InMemory:    true,
+					StickyVFSID: strconv.FormatInt(int64(i), 10),
+				},
+			},
+			Knobs: base.TestingKnobs{
+				Server: &server.TestingKnobs{
+					StickyVFSRegistry: stickyVFSRegistry,
+				},
+			},
+		}
+	}
+
+	tc := testcluster.StartTestCluster(t, numServers,
+		base.TestClusterArgs{
+			ReplicationMode:     base.ReplicationManual,
+			ReusableListenerReg: lisReg,
+			ServerArgsPerNode:   stickyServerArgs,
+		})
+
+	defer tc.Stopper().Stop(ctx)
+
+	key := tc.ScratchRange(t)
+	tc.AddVotersOrFatal(t, key, tc.Targets(1)...)
+	desc, err := tc.LookupRange(key)
+	require.NoError(t, err)
+
+	// Randomly stop server index 0 or 1.
+	stoppedNodeInx := rand.Intn(2)
+	aliveNodeIdx := 1 - stoppedNodeInx
+	tc.StopServer(stoppedNodeInx)
+	repl := tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+
+	// The range is available initially.
+	require.False(t, repl.LeaderlessWatcher.IsUnavailable())
+
+	// Wait until the remaining replica becomes leaderless.
+	testutils.SucceedsSoon(t, func() error {
+		if repl.RaftStatus().Lead != raft.None {
+			return errors.New("Leader still exists")
+		}
+		return nil
+	})
+
+	// Increment the clock by the leaderlessWatcher unavailable threshold.
+	manualClock.Increment(threshold.Nanoseconds())
+
+	// Wait for the leaderlessWatcher to indicate that the range is unavailable.
+	testutils.SucceedsSoon(t, func() error {
+		tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+		if !repl.LeaderlessWatcher.IsUnavailable() {
+			return errors.New("range is still available")
+		}
+		return nil
+	})
+
+	sendPutRequestWithTimeout := func(repl *kvserver.Replica, timeout time.Duration) (*kvpb.Error, error) {
+		ctx, cancel := context.WithTimeout(ctx, timeout)
+		defer cancel()
+		ba := &kvpb.BatchRequest{}
+		ba.RangeID = desc.RangeID
+		ba.Timestamp = repl.Clock().Now()
+		ba.Add(putArgs(key, []byte("foo")))
+		_, pErr := repl.Send(ctx, ba)
+		return pErr, ctx.Err()
+	}
+
+	// Requests should immediately return an error indicating that the range is
+	// unavailable.
+	pErr, ctxErr := sendPutRequestWithTimeout(repl, 2*time.Second)
+	require.NoError(t, ctxErr)
+	require.Regexp(t, "replica has been leaderless for 10s", pErr)
+	require.True(t, errors.HasType(pErr.GoError(), (*kvpb.ReplicaUnavailableError)(nil)),
+		"expected ReplicaUnavailableError, got %v", err)
+
+	// At this point we know that the replica is considered unavailable. Regain
+	// the quorum and check that the leaderlessWatcher indicates that the range is
+	// available.
+	require.NoError(t, tc.RestartServer(stoppedNodeInx))
+
+	testutils.SucceedsSoon(t, func() error {
+		repl = tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+		tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+		if repl.LeaderlessWatcher.IsUnavailable() {
+			return errors.New("range is still unavailable")
+		}
+		return nil
+	})
+
+	// Requests should now succeed. We need to try both replicas to avoid
+	// NotLeaseHolderErrors.
+	testutils.SucceedsSoon(t, func() error {
+		for i := range numServers {
+			repl := tc.GetFirstStoreFromServer(t, i).LookupReplica(roachpb.RKey(key))
+			pErr, ctxErr = sendPutRequestWithTimeout(repl, 2*time.Second)
+			if ctxErr == nil && pErr == nil {
+				return nil
+			}
+		}
+		// If we reach this point, we know that the request failed, return the
+		// error.
+		if ctxErr != nil {
+			return ctxErr
+		}
+		return pErr.GoError()
+	})
+}
+
 func TestClearRange(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/kv/kvserver/replica.go

@@ -232,6 +232,11 @@ func (lw *leaderlessWatcher) IsUnavailable() bool {
 	return lw.mu.unavailable
 }
 
+func (lw *leaderlessWatcher) resetLocked() {
+	lw.mu.leaderlessTimestamp = time.Time{}
+	lw.mu.unavailable = false
+}
+
 // 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

pkg/kv/kvserver/replica_backpressure.go

@@ -118,18 +118,29 @@ func canBackpressureBatch(ba *kvpb.BatchRequest) bool {
 	return false
 }
 
-// signallerForBatch returns the signaller to use for this batch. This is the
-// Replica's breaker's signaller except if any request in the batch uses
-// poison.Policy_Wait, in which case it's a neverTripSignaller. In particular,
-// `(signaller).C() == nil` signals that the request bypasses the circuit
-// breakers.
+// signallerForBatch returns the signaller to use for this batch in the
+// following priorities:
+// 1. If the batch contains a request uses poison.Policy_Wait, we will return
+// neverTripSignaller.
+// 2. If the replica is leaderless for a time longer than the threshold in
+// `kv.replica_raft.leaderless_unavailable_threshold`, use the
+// leaderlessWatcher signal.
+// 3. Otherwise, use the replica's breaker's signaller
 func (r *Replica) signallerForBatch(ba *kvpb.BatchRequest) signaller {
 	for _, ru := range ba.Requests {
 		req := ru.GetInner()
 		if kvpb.BypassesReplicaCircuitBreaker(req) {
 			return neverTripSignaller{}
 		}
 	}
+
+	// If the leaderless watcher indicates that this replica is leaderless for a
+	// long time, use it as the signal.
+	if r.LeaderlessWatcher.IsUnavailable() {
+		return r.LeaderlessWatcher
+	}
+
+	// Otherwise, use the replica's breaker.
 	return r.breaker.Signal()
 }
 

pkg/kv/kvserver/replica_raft.go

@@ -1540,19 +1540,25 @@ func (r *Replica) tick(
 	//
 	// This is likely unintentional, and the leader should likely consider itself
 	// live even when quiesced.
+	nowPhysicalTime := r.Clock().PhysicalTime()
 	if r.isRaftLeaderRLocked() {
-		r.mu.lastUpdateTimes.update(r.replicaID, r.Clock().PhysicalTime())
+		r.mu.lastUpdateTimes.update(r.replicaID, nowPhysicalTime)
 		// We also update lastUpdateTimes for replicas that provide store liveness
 		// support to the leader.
 		r.updateLastUpdateTimesUsingStoreLivenessRLocked(storeClockTimestamp)
 	}
 
 	r.mu.ticks++
-	preTickState := r.mu.internalRaftGroup.BasicStatus().RaftState
+	preTickStatus := r.mu.internalRaftGroup.BasicStatus()
 	r.mu.internalRaftGroup.Tick()
-	postTickState := r.mu.internalRaftGroup.BasicStatus().RaftState
-	if preTickState != postTickState {
-		if postTickState == raftpb.StatePreCandidate {
+	postTickStatus := r.mu.internalRaftGroup.BasicStatus()
+
+	// Check if the replica has been leaderless for too long, and potentially set
+	// the leaderless watcher replica state as unavailable.
+	r.maybeMarkReplicaUnavailableInLeaderlessWatcher(ctx, postTickStatus.Lead, nowPhysicalTime)
+
+	if preTickStatus.RaftState != postTickStatus.RaftState {
+		if postTickStatus.RaftState == raftpb.StatePreCandidate {
 			r.store.Metrics().RaftTimeoutCampaign.Inc(1)
 			if k := r.store.TestingKnobs(); k != nil && k.OnRaftTimeoutCampaign != nil {
 				k.OnRaftTimeoutCampaign(r.RangeID)
@@ -2179,6 +2185,47 @@ func (r *Replica) reportSnapshotStatus(ctx context.Context, to roachpb.ReplicaID
 	}
 }
 
+// maybeMarkReplicaUnavailableInLeaderlessWatcher marks the replica as
+// unavailable in the leaderless watcher if the replica has been leaderless
+// for a duration of time greater than or equal to the threshold.
+func (r *Replica) maybeMarkReplicaUnavailableInLeaderlessWatcher(
+	ctx context.Context, postTickLead raftpb.PeerID, storeClockTime time.Time,
+) {
+	r.LeaderlessWatcher.mu.Lock()
+	defer r.LeaderlessWatcher.mu.Unlock()
+
+	threshold := ReplicaLeaderlessUnavailableThreshold.Get(&r.store.cfg.Settings.SV)
+	if threshold == time.Duration(0) {
+		// The leaderless watcher is disabled. It's important to reset the
+		// leaderless watcher when it's disabled to reset any replica that was
+		// marked as unavailable before the watcher was disabled.
+		r.LeaderlessWatcher.resetLocked()
+		return
+	}
+
+	if postTickLead != raft.None {
+		// If we know about the leader, reset the leaderless timer, and mark the
+		// replica as available.
+		r.LeaderlessWatcher.resetLocked()
+	} else if r.LeaderlessWatcher.mu.leaderlessTimestamp.IsZero() {
+		// If we don't know about the leader, and we haven't been leaderless before,
+		// mark the time we became leaderless.
+		r.LeaderlessWatcher.mu.leaderlessTimestamp = storeClockTime
+	} else if !r.LeaderlessWatcher.mu.unavailable {
+		// At this point we know that we have been leaderless for sometime, and
+		// we haven't marked the replica as unavailable yet. Make sure we didn't
+		// exceed the threshold. Otherwise, mark the replica as unavailable.
+		durationSinceLeaderless := storeClockTime.Sub(r.LeaderlessWatcher.mu.leaderlessTimestamp)
+		if durationSinceLeaderless >= threshold {
+			err := errors.Errorf("have been leaderless for %.2fs, setting the "+
+				"leaderless watcher replica's state as unavailable",
+				durationSinceLeaderless.Seconds())
+			log.Warningf(ctx, "%s", err)
+			r.LeaderlessWatcher.mu.unavailable = true
+		}
+	}
+}
+
 type snapTruncationInfo struct {
 	index          kvpb.RaftIndex
 	recipientStore roachpb.StoreID

pkg/kv/kvserver/replica_raft_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/logstore"
+	"github.com/cockroachdb/cockroach/pkg/raft"
 	"github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/raft/tracker"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -386,3 +387,114 @@ func checkNoLeakedTraceSpans(t *testing.T, store *Store) {
 		return errors.Newf("%s\n\ngoroutines of interest: %v\nstacks:\n\n%s", buf.String(), ids, sl)
 	})
 }
+
+// TestMaybeMarkReplicaUnavailableInLeaderlessWatcher is a basic unit test for
+// the function maybeMarkReplicaUnavailableInLeaderlessWatcher.
+func TestMaybeMarkReplicaUnavailableInLeaderlessWatcher(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	now := time.Now()
+	leaderlessThreshold := time.Second * 60
+
+	testCases := []struct {
+		name                   string
+		initReplicaUnavailable bool
+		// The initial leaderless timestamp of the replica in the leaderlessWatcher.
+		initLeaderlessTimestamp time.Time
+		leader                  raftpb.PeerID
+		disableWatcher          bool
+		expectedLeaderlessTime  time.Time
+		expectedUnavailable     bool
+	}{
+		{
+			name:                    "leader known",
+			initLeaderlessTimestamp: time.Time{},
+			leader:                  raftpb.PeerID(1),
+			disableWatcher:          false,
+			// Since the leader is known, we expect that the replica is considered
+			// available, and the leaderless timestamp is reset.
+			expectedLeaderlessTime: time.Time{},
+			expectedUnavailable:    false,
+		},
+		{
+			name:                    "leader was unknown but is now known",
+			initLeaderlessTimestamp: now.Add(-10 * time.Second),
+			leader:                  raftpb.PeerID(1),
+			disableWatcher:          false,
+			// Since the leader is known, we expect that the replica is considered
+			// available, and the leaderless timestamp is reset.
+			expectedLeaderlessTime: time.Time{},
+			expectedUnavailable:    false,
+		},
+		{
+			name:                    "leader unknown less than threshold",
+			initLeaderlessTimestamp: now.Add(-10 * time.Second),
+			leader:                  raft.None,
+			disableWatcher:          false,
+			// Since the leader has been unknown for less than the threshold, we
+			// expect that the replica is considered available, and the leaderless
+			// time it is set properly.
+			expectedLeaderlessTime: now.Add(-10 * time.Second),
+			expectedUnavailable:    false,
+		},
+		{
+			name:                    "leader unknown exceeds threshold",
+			initLeaderlessTimestamp: now.Add(-leaderlessThreshold),
+			leader:                  raft.None,
+			disableWatcher:          false,
+			// Since the leader has been unknown for the threshold period, we
+			// expect that the replica is considered unavailable, and the leaderless
+			// time it is set properly.
+			expectedLeaderlessTime: now.Add(-leaderlessThreshold),
+			expectedUnavailable:    true,
+		},
+		{
+			name:                    "leader unknown exceeds threshold and watcher disabled",
+			initReplicaUnavailable:  true,
+			initLeaderlessTimestamp: now.Add(-leaderlessThreshold),
+			leader:                  raft.None,
+			disableWatcher:          true,
+			// Since the watcher is disabled, the replica won't be considered
+			// unavailable even if the replica was marked as unavailable, and it's
+			// been leaderless for a long time.
+			expectedLeaderlessTime: time.Time{},
+			expectedUnavailable:    false,
+		},
+	}
+	for _, tc := range testCases {
+		ctx := context.Background()
+		stopper := stop.NewStopper()
+
+		tContext := testContext{}
+		cfg := TestStoreConfig(nil)
+
+		if tc.disableWatcher {
+			ReplicaLeaderlessUnavailableThreshold.Override(ctx, &cfg.Settings.SV, time.Duration(0))
+		} else {
+			ReplicaLeaderlessUnavailableThreshold.Override(ctx, &cfg.Settings.SV, leaderlessThreshold)
+		}
+		tContext.StartWithStoreConfig(ctx, t, stopper, cfg)
+
+		repl := tContext.repl
+		repl.LeaderlessWatcher.mu.unavailable = tc.initReplicaUnavailable
+		repl.LeaderlessWatcher.mu.leaderlessTimestamp = tc.initLeaderlessTimestamp
+		repl.maybeMarkReplicaUnavailableInLeaderlessWatcher(ctx, tc.leader, now)
+		require.Equal(t, tc.expectedUnavailable, repl.LeaderlessWatcher.IsUnavailable())
+		require.Equal(t, tc.expectedLeaderlessTime, repl.LeaderlessWatcher.mu.leaderlessTimestamp)
+
+		// Attempt to write to the replica to ensure that if it's considered
+		// unavailable, we should get an error.
+		key := roachpb.Key("a")
+		write := putArgs(key, []byte("foo"))
+		_, pErr := tContext.SendWrapped(&write)
+
+		if tc.expectedUnavailable {
+			require.Regexp(t, "replica has been leaderless for 1m0s", pErr)
+		} else {
+			require.NoError(t, pErr.GoError())
+		}
+
+		stopper.Stop(ctx)
+	}
+}