kvserver: refresh ReplicaUnavailableErrors on the leaderlessWatcher

Previously, we would construct a ReplicaUnavailableError on the
LeaderlessWatcher once (using an empty descriptor, which is rather
hilarious) and never updated the error. That meant that the returned
error wouldn't reflect the current state of the RangeDescriptor, which
made it hard to make sense of the shape of the unavailability.

This patch fixes the issue by refreshing the cached error on every
available -> unavailable state transition in the leaderlessWatcher.

Closes #144639

Release note: None

Author: arulajmani

pkg/kv/kvserver/client_replica_test.go

@@ -68,6 +68,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
+	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
@@ -2788,7 +2789,6 @@ func TestLossQuorumCauseLeaderWatcherToSignalUnavailable(t *testing.T) {
 	require.NoError(t, log.SetVModule("replica_range_lease=3,raft=4"))
 
 	ctx := context.Background()
-	manualClock := hlc.NewHybridManualClock()
 	stickyVFSRegistry := fs.NewStickyRegistry()
 	lisReg := listenerutil.NewListenerRegistry()
 	defer lisReg.Close()
@@ -2851,12 +2851,8 @@ func TestLossQuorumCauseLeaderWatcherToSignalUnavailable(t *testing.T) {
 		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")
 		}
@@ -2915,6 +2911,77 @@ func TestLossQuorumCauseLeaderWatcherToSignalUnavailable(t *testing.T) {
 	})
 }
 
+// TestLeaderlessWatcherUnavailabilityErrorRefreshedOnUnavailabilityTransition
+// ensures that the leaderless watcher constructs a new error every time it
+// transitions to the unavailable state. In particular, the descriptor used
+// in the error should be the latest descriptor.
+// Serves as a regression test for
+// https://github.com/cockroachdb/cockroach/issues/144639.
+func TestLeaderlessWatcherErrorRefreshedOnUnavailabilityTransition(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+
+	manual := hlc.NewHybridManualClock()
+	st := cluster.MakeTestingClusterSettings()
+	// Set the leaderless threshold to 10 second.
+	kvserver.ReplicaLeaderlessUnavailableThreshold.Override(ctx, &st.SV, 10*time.Second)
+
+	tc := testcluster.StartTestCluster(t, 3, base.TestClusterArgs{
+		ReplicationMode: base.ReplicationManual,
+		ServerArgs: base.TestServerArgs{
+			Settings: st,
+			Knobs: base.TestingKnobs{
+				Server: &server.TestingKnobs{
+					WallClock: manual,
+				},
+			},
+		},
+	})
+	defer tc.Stopper().Stop(ctx)
+	key := tc.ScratchRange(t)
+	tc.AddVotersOrFatal(t, key, tc.Targets(1)...)
+	repl := tc.GetFirstStoreFromServer(t, 1).LookupReplica(roachpb.RKey(key))
+
+	// The leaderlessWatcher starts off as available.
+	require.False(t, repl.LeaderlessWatcher.IsUnavailable())
+	// Let it know it's leaderless.
+	repl.RefreshLeaderlessWatcherUnavailableStateForTesting(ctx, raft.None, manual.Now(), st)
+	// Even though the replica is leaderless, enough time hasn't passed for it to
+	// be considered unavailable.
+	require.False(t, repl.LeaderlessWatcher.IsUnavailable())
+	// The error should be nil as we're not considered leaderless at this point.
+	require.NoError(t, repl.LeaderlessWatcher.Err())
+	// Let enough time pass.
+	manual.Increment(10 * time.Second.Nanoseconds())
+	repl.RefreshLeaderlessWatcherUnavailableStateForTesting(ctx, raft.None, manual.Now(), st)
+	// Now the replica is considered unavailable.
+	require.True(t, repl.LeaderlessWatcher.IsUnavailable())
+	require.Error(t, repl.LeaderlessWatcher.Err())
+	// Regex to ensure we've got a replica unavailable error with n1 and n2 in the
+	// range descriptor.
+	require.Regexp(t, "replica unavailable.*n1.*n2.*gen=3", repl.LeaderlessWatcher.Err().Error())
+
+	// Next up, let the replica know there's a leader. This should make it
+	// available again.
+	repl.RefreshLeaderlessWatcherUnavailableStateForTesting(ctx, 1, manual.Now(), st)
+	require.False(t, repl.LeaderlessWatcher.IsUnavailable())
+	// Change the range descriptor. Mark it leaderless and let enough time pass
+	// for it to be considered unavailable again.
+	tc.AddVotersOrFatal(t, key, tc.Targets(2)...)
+	repl.RefreshLeaderlessWatcherUnavailableStateForTesting(ctx, raft.None, manual.Now(), st)
+	manual.Increment(10 * time.Second.Nanoseconds())
+	repl.RefreshLeaderlessWatcherUnavailableStateForTesting(ctx, raft.None, manual.Now(), st)
+	// The replica should now be considered unavailable again.
+	require.True(t, repl.LeaderlessWatcher.IsUnavailable())
+	require.Error(t, repl.LeaderlessWatcher.Err())
+	// Ensure that the range descriptor now contains n1, n2, and n3 -- i.e, we're
+	// updating the error with the latest descriptor on the latest transition.
+	require.Regexp(t, "replica unavailable.*n1.*n2.*n3.*gen=5", repl.LeaderlessWatcher.Err().Error())
+}
+
 func TestClearRange(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/kv/kvserver/replica.go

@@ -195,10 +195,10 @@ type leaderlessWatcher struct {
 		// 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
+		// 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
@@ -208,22 +208,21 @@ type leaderlessWatcher struct {
 	closedChannel chan struct{}
 }
 
-// newLeaderlessWatcher initializes a new leaderlessWatcher with the default
-// values.
-func newLeaderlessWatcher(r *Replica) *leaderlessWatcher {
+// newLeaderlessWatcher constructs and returns a new leaderlessWatcher.
+func newLeaderlessWatcher() *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,
 	}
 }
 
 // Err implements the signaller interface.
 func (lw *leaderlessWatcher) Err() error {
-	return lw.err
+	lw.mu.RLock()
+	defer lw.mu.RUnlock()
+
+	return lw.mu.err
 }
 
 // C implements the signaller interface.
@@ -237,14 +236,23 @@ func (lw *leaderlessWatcher) C() <-chan struct{} {
 func (lw *leaderlessWatcher) IsUnavailable() bool {
 	lw.mu.RLock()
 	defer lw.mu.RUnlock()
+
+	// The error is set iff the replica is unavailable. Sanity check.
+	if lw.mu.unavailable == (lw.mu.err == nil) {
+		panic("unavailable implies error is set")
+	}
 	return lw.mu.unavailable
 }
 
 // refreshUnavailableState refreshes the unavailable state on the leaderless
 // watcher. Replicas are considered unavailable if they have been leaderless for
 // a long time, where long is defined by the ReplicaUnavailableThreshold.
 func (lw *leaderlessWatcher) refreshUnavailableState(
-	ctx context.Context, postTickLead raftpb.PeerID, nowPhysicalTime time.Time, st *cluster.Settings,
+	ctx context.Context,
+	postTickLead raftpb.PeerID,
+	nowPhysicalTime time.Time,
+	st *cluster.Settings,
+	newReplicaUnavailableError func(error) error,
 ) {
 	lw.mu.Lock()
 	defer lw.mu.Unlock()
@@ -272,20 +280,28 @@ func (lw *leaderlessWatcher) refreshUnavailableState(
 		// the threshold. Otherwise, mark the replica as unavailable.
 		durationSinceLeaderless := nowPhysicalTime.Sub(lw.mu.leaderlessTimestamp)
 		if durationSinceLeaderless >= threshold {
-			err := errors.Errorf("have been leaderless for %.2fs, setting the "+
-				"leaderless watcher replica's state as unavailable",
-				durationSinceLeaderless.Seconds())
 			if log.ExpensiveLogEnabled(ctx, 1) {
+				err := errors.Errorf("have been leaderless for %.2fs, setting the "+
+					"leaderless watcher replica's state as unavailable",
+					durationSinceLeaderless.Seconds())
 				log.VEventf(ctx, 1, "%s", err)
 			}
+			// Transition to being unavailable.
 			lw.mu.unavailable = true
+			// Now that we're transitioning to being unavailable, construct and cache
+			// the associated error.
+			lw.mu.err = newReplicaUnavailableError(
+				errors.Errorf("replica has been leaderless for %s",
+					ReplicaLeaderlessUnavailableThreshold.Get(&st.SV)),
+			)
 		}
 	}
 }
 
 func (lw *leaderlessWatcher) resetLocked() {
 	lw.mu.leaderlessTimestamp = time.Time{}
 	lw.mu.unavailable = false
+	lw.mu.err = nil
 }
 
 // ReplicaMutex is an RWMutex. It has its own type to make it easier to look for
@@ -2822,6 +2838,17 @@ func (r *Replica) GetCachedClosedTimestampPolicyForTesting() ctpb.RangeClosedTim
 	return ctpb.RangeClosedTimestampPolicy(r.cachedClosedTimestampPolicy.Load())
 }
 
+// RefreshLeaderlessWatcherUnavailableStateForTesting refreshes the replica's
+// leaderlessWatcher's unavailable state. Intended for tests.
+func (r *Replica) RefreshLeaderlessWatcherUnavailableStateForTesting(
+	ctx context.Context, postTickLead raftpb.PeerID, nowPhysicalTime time.Time, st *cluster.Settings,
+) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	r.LeaderlessWatcher.refreshUnavailableState(ctx, postTickLead, nowPhysicalTime, st, r.replicaUnavailableErrorRLocked)
+}
+
 // maybeEnqueueProblemRange will enqueue the replica for processing into the
 // replicate queue iff:
 //

pkg/kv/kvserver/replica_circuit_breaker.go

@@ -268,11 +268,23 @@ func replicaUnavailableError(
 	return kvpb.NewReplicaUnavailableError(errors.Wrapf(err, "%s", buf), desc, replDesc)
 }
 
+// replicaUnavailableError returns a new ReplicaUnavailableError that wraps the
+// provided error.
 func (r *Replica) replicaUnavailableError(err error) error {
-	desc := r.Desc()
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	return r.replicaUnavailableErrorRLocked(err)
+}
+
+// replicaUnavailableLocked is like replicaUnavailableError, except it requires
+// r.mu to be RLocked.
+func (r *Replica) replicaUnavailableErrorRLocked(err error) error {
+	desc := r.shMu.state.Desc
 	replDesc, _ := desc.GetReplicaDescriptor(r.store.StoreID())
 
 	isLiveMap, _ := r.store.livenessMap.Load().(livenesspb.IsLiveMap)
-	ct := r.GetCurrentClosedTimestamp(context.Background())
-	return replicaUnavailableError(err, desc, replDesc, isLiveMap, r.RaftStatus(), ct)
+	ct := r.getCurrentClosedTimestampLocked(context.Background(), hlc.Timestamp{} /* sufficient */)
+
+	return replicaUnavailableError(err, desc, replDesc, isLiveMap, r.raftStatusRLocked(), ct)
 }

pkg/kv/kvserver/replica_init.go

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

pkg/kv/kvserver/replica_raft.go

@@ -1512,7 +1512,7 @@ func (r *Replica) tick(
 
 	// Refresh the unavailability state on the leaderlessWatcher.
 	r.LeaderlessWatcher.refreshUnavailableState(
-		ctx, postTickStatus.Lead, nowPhysicalTime, r.store.cfg.Settings,
+		ctx, postTickStatus.Lead, nowPhysicalTime, r.store.cfg.Settings, r.replicaUnavailableErrorRLocked,
 	)
 
 	if preTickStatus.RaftState != postTickStatus.RaftState {

pkg/kv/kvserver/replica_raft_test.go

@@ -481,7 +481,7 @@ func TestMaybeMarkReplicaUnavailableInLeaderlessWatcher(t *testing.T) {
 		repl := tContext.repl
 		repl.LeaderlessWatcher.mu.unavailable = tc.initReplicaUnavailable
 		repl.LeaderlessWatcher.mu.leaderlessTimestamp = tc.initLeaderlessTimestamp
-		repl.LeaderlessWatcher.refreshUnavailableState(ctx, tc.leader, now, cfg.Settings)
+		repl.RefreshLeaderlessWatcherUnavailableStateForTesting(ctx, tc.leader, now, cfg.Settings)
 		require.Equal(t, tc.expectedUnavailable, repl.LeaderlessWatcher.IsUnavailable())
 		require.Equal(t, tc.expectedLeaderlessTime, repl.LeaderlessWatcher.mu.leaderlessTimestamp)
 

pkg/kv/kvserver/replica_test.go

@@ -15429,8 +15429,8 @@ func TestLeaderlessWatcherInit(t *testing.T) {
 	// 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 error is nilled out.
+	require.Nil(t, repl.LeaderlessWatcher.mu.err)
 
 	// The channel is closed.
 	c := repl.LeaderlessWatcher.C()