kvserver: support atomic promotions and demotions of non-voting replicas

This PR teaches `AdminChangeReplicas` to atomically promote voters to
non-voters, demote voters to non-voters or swap voters with non-voters
via joint consensus.

Fixes cockroachdb#58499
Informs cockroachdb#51943

Release note: None

pkg/kv/kvserver/allocator.go

@@ -614,7 +614,7 @@ func (a Allocator) simulateRemoveTarget(
 		a.storePool.updateLocalStoreAfterRebalance(targetStore, rangeUsageInfo, roachpb.REMOVE_VOTER)
 	}()
 	log.VEventf(ctx, 3, "simulating which replica would be removed after adding s%d", targetStore)
-	return a.RemoveTarget(ctx, zone, candidates, existingReplicas)
+	return a.RemoveTarget(ctx, zone, roachpb.MakeReplicaSet(candidates).ReplicationTargets(), existingReplicas)
 }
 
 // RemoveTarget returns a suitable replica to remove from the provided replica
@@ -625,7 +625,7 @@ func (a Allocator) simulateRemoveTarget(
 func (a Allocator) RemoveTarget(
 	ctx context.Context,
 	zone *zonepb.ZoneConfig,
-	candidates []roachpb.ReplicaDescriptor,
+	candidates []roachpb.ReplicationTarget,
 	existingReplicas []roachpb.ReplicaDescriptor,
 ) (roachpb.ReplicaDescriptor, string, error) {
 	if len(candidates) == 0 {
@@ -923,7 +923,7 @@ func (a *Allocator) TransferLeaseTarget(
 		// If the current leaseholder is not preferred, set checkTransferLeaseSource
 		// to false to motivate the below logic to transfer the lease.
 		existing = preferred
-		if !storeHasReplica(leaseStoreID, preferred) {
+		if !storeHasReplica(leaseStoreID, roachpb.MakeReplicaSet(preferred).ReplicationTargets()) {
 			checkTransferLeaseSource = false
 		}
 	}
@@ -1023,7 +1023,7 @@ func (a *Allocator) ShouldTransferLease(
 		existing = preferred
 		// If the current leaseholder isn't one of the preferred stores, then we
 		// should try to transfer the lease.
-		if !storeHasReplica(leaseStoreID, existing) {
+		if !storeHasReplica(leaseStoreID, roachpb.MakeReplicaSet(existing).ReplicationTargets()) {
 			return true
 		}
 	}

pkg/kv/kvserver/allocator_scorer.go

@@ -882,7 +882,7 @@ func nodeHasReplica(nodeID roachpb.NodeID, existing []roachpb.ReplicaDescriptor)
 
 // storeHasReplica returns true if the provided StoreID contains an entry in
 // the provided list of existing replicas.
-func storeHasReplica(storeID roachpb.StoreID, existing []roachpb.ReplicaDescriptor) bool {
+func storeHasReplica(storeID roachpb.StoreID, existing []roachpb.ReplicationTarget) bool {
 	for _, r := range existing {
 		if r.StoreID == storeID {
 			return true

pkg/kv/kvserver/allocator_scorer_test.go

@@ -418,7 +418,8 @@ func TestStoreHasReplica(t *testing.T) {
 		existing = append(existing, roachpb.ReplicaDescriptor{StoreID: roachpb.StoreID(i)})
 	}
 	for i := 1; i < 10; i++ {
-		if e, a := i%2 == 0, storeHasReplica(roachpb.StoreID(i), existing); e != a {
+		if e, a := i%2 == 0,
+			storeHasReplica(roachpb.StoreID(i), roachpb.MakeReplicaSet(existing).ReplicationTargets()); e != a {
 			t.Errorf("StoreID %d expected to be %t, got %t", i, e, a)
 		}
 	}

pkg/kv/kvserver/allocator_test.go

@@ -2111,7 +2111,7 @@ func TestAllocatorRemoveTargetLocality(t *testing.T) {
 		targetRepl, details, err := a.RemoveTarget(
 			context.Background(),
 			zonepb.EmptyCompleteZoneConfig(),
-			existingRepls,
+			roachpb.MakeReplicaSet(existingRepls).ReplicationTargets(),
 			existingRepls,
 		)
 		if err != nil {
@@ -4214,7 +4214,7 @@ func TestAllocatorRemoveTarget(t *testing.T) {
 		targetRepl, _, err := a.RemoveTarget(
 			ctx,
 			zonepb.EmptyCompleteZoneConfig(),
-			replicas,
+			roachpb.MakeReplicaSet(replicas).ReplicationTargets(),
 			replicas,
 		)
 		if err != nil {

pkg/kv/kvserver/client_lease_test.go

@@ -350,7 +350,7 @@ func TestCannotTransferLeaseToVoterOutgoing(t *testing.T) {
 					// We have a sleep loop below to try to encourage the lease transfer
 					// to make it past that sanity check prior to letting the change
 					// of replicas proceed.
-					"cannot transfer lease to replica of type VOTER_DEMOTING", err.Error())
+					"cannot transfer lease to replica of type VOTER_DEMOTING_LEARNER", err.Error())
 		}()
 		// Try really hard to make sure that our request makes it past the
 		// sanity check error to the evaluation error.

pkg/kv/kvserver/client_replica_test.go

@@ -2289,6 +2289,37 @@ func TestRandomConcurrentAdminChangeReplicasRequests(t *testing.T) {
 	}
 }
 
+func TestChangeReplicasSwapVoterWithNonVoter(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	const numNodes = 7
+	tc := testcluster.StartTestCluster(t, numNodes, base.TestClusterArgs{
+		ReplicationMode: base.ReplicationManual,
+	})
+	ctx := context.Background()
+	defer tc.Stopper().Stop(ctx)
+
+	key := tc.ScratchRange(t)
+	// NB: The test cluster starts with firstVoter having a voting replica (and
+	// the lease) for all ranges.
+	firstVoter, secondVoter, nonVoter := tc.Target(0), tc.Target(1), tc.Target(3)
+	firstStore, err := tc.Server(0).GetStores().(*kvserver.Stores).GetStore(tc.Server(0).GetFirstStoreID())
+	require.NoError(t, err)
+	firstRepl := firstStore.LookupReplica(roachpb.RKey(key))
+	require.NotNil(t, firstRepl, `the first node in the TestCluster must have a replica for the ScratchRange`)
+
+	// TODO(aayush): Trying to swap the last voting replica with a non-voter hits
+	// the safeguard inside Replica.propose() as the last voting replica is always
+	// the leaseholder. There are a bunch of subtleties around getting a
+	// leaseholder to remove itself without another voter to immediately transfer
+	// the lease to. Determine if/how this needs to be fixed.
+	tc.AddNonVotersOrFatal(t, key, nonVoter)
+	_, err = tc.SwapVoterWithNonVoter(key, firstVoter, nonVoter)
+	require.Regexp(t, "received invalid ChangeReplicasTrigger", err)
+
+	tc.AddVotersOrFatal(t, key, secondVoter)
+	tc.SwapVoterWithNonVoterOrFatal(t, key, secondVoter, nonVoter)
+}
+
 // TestReplicaTombstone ensures that tombstones are written when we expect
 // them to be. Tombstones are laid down when replicas are removed.
 // Replicas are removed for several reasons:

pkg/kv/kvserver/raft.proto

@@ -147,7 +147,7 @@ message SnapshotRequest {
     // for upreplication, before they're promoted to full voters.
     //
     // As of the time of writing, we only send this snapshot from the
-    // atomicReplicationChange after creating a new LEARNER replica.
+    // execReplicationChangesForVoters after creating a new LEARNER replica.
     LEARNER_INITIAL = 1;
     reserved 2;
   }