kvserver: remove changed replicas in purgatory from replica set

It was possible for a replica to be stuck processing in a queue's
replica set. This could occur when a replica had recently been removed
from purgatory for processing but was destroyed, or replica ID changed
before being processed.

When this occurred, the replica could never be processed by the queue
again, potentially leading to decommission stalls, constraint violations
or under(over)replication.

Remove the replica from the queue set upon encountering a replica which
was destroyed, or replica ID changed when processing purgatory. This
prevents the replica from becoming stuck in a processing state in the
queue set.

Fixes: #112761
Fixes: #110761

Release note (bug fix): The store queues will no longer leave purgatory
replicas which have changed replica IDs, or have been destroyed stuck
unable to process via the respective queue again if re-added.

Author: kvoli

pkg/kv/kvserver/queue.go

@@ -1272,6 +1272,9 @@ func (bq *baseQueue) processReplicasInPurgatory(
 		for _, item := range ranges {
 			repl, err := bq.getReplica(item.rangeID)
 			if err != nil || item.replicaID != repl.ReplicaID() {
+				bq.mu.Lock()
+				bq.removeFromReplicaSetLocked(item.rangeID)
+				bq.mu.Unlock()
 				continue
 			}
 			annotatedCtx := repl.AnnotateCtx(ctx)
@@ -1281,6 +1284,10 @@ func (bq *baseQueue) processReplicasInPurgatory(
 					bq.finishProcessingReplica(ctx, stopper, repl, err)
 				},
 			) != nil {
+				// NB: We do not need to worry about removing any unprocessed replicas
+				// from the replica set here, as RunTask will only return an error when
+				// the stopper is quiescing or stopping -- meaning the process is
+				// shutting down.
 				return
 			}
 		}

pkg/kv/kvserver/queue_test.go

@@ -904,17 +904,28 @@ func TestBaseQueuePurgatory(t *testing.T) {
 		return nil
 	})
 
+	// Change the replicaID of the first the replica and destroy the second
+	// replica. These replicas should not be processed and should be removed from
+	// the replica set. The number of processed replicas will be 2 less.
+	const rmReplCount = 2
+	repls[0].replicaID = 2
+	if err := tc.store.RemoveReplica(ctx, repls[1], repls[1].Desc().NextReplicaID, RemoveOptions{
+		DestroyData: true,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
 	// Remove error and reprocess.
 	testQueue.err = nil
 	testQueue.pChan <- timeutil.Now()
 
 	testutils.SucceedsSoon(t, func() error {
-		if pc := testQueue.getProcessed(); pc != replicaCount*3 {
-			return errors.Errorf("expected %d processed replicas; got %d", replicaCount*3, pc)
+		if pc := testQueue.getProcessed(); pc != replicaCount*3-rmReplCount {
+			return errors.Errorf("expected %d processed replicas; got %d", replicaCount*3-rmReplCount, pc)
 		}
 		// Check metrics.
-		if v := bq.successes.Count(); v != int64(replicaCount) {
-			return errors.Errorf("expected %d processed replicas; got %d", replicaCount, v)
+		if v := bq.successes.Count(); v != int64(replicaCount)-rmReplCount {
+			return errors.Errorf("expected %d processed replicas; got %d", replicaCount-rmReplCount, v)
 		}
 		if v := bq.failures.Count(); v != int64(replicaCount*2) {
 			return errors.Errorf("expected %d failed replicas; got %d", replicaCount*2, v)
@@ -925,6 +936,15 @@ func TestBaseQueuePurgatory(t *testing.T) {
 		if v := bq.purgatory.Value(); v != 0 {
 			return errors.Errorf("expected 0 purgatory replicas; got %d", v)
 		}
+		// Verify there are no replicas left in the replica set after finishing
+		// processing. This is within the retry loop as the above conditions can
+		// pass without considering the removed replicas.
+		bq.mu.Lock()
+		replicasCount := len(bq.mu.replicas)
+		bq.mu.Unlock()
+		if replicasCount != 0 {
+			return errors.Errorf("expected no replicas in the replica set: got %d", replicasCount)
+		}
 		return nil
 	})
 
@@ -936,6 +956,30 @@ func TestBaseQueuePurgatory(t *testing.T) {
 	if l := bq.Length(); l != 0 {
 		t.Errorf("expected empty priorityQ; got %d", l)
 	}
+
+	// Verify that the replica with a changed replicaID can be processed.
+	beforeProcessCount := testQueue.getProcessed()
+	beforeSuccessCount := bq.successes.Count()
+	beforeFailureCount := bq.failures.Count()
+	bq.maybeAdd(ctx, repls[0], hlc.ClockTimestamp{})
+	testutils.SucceedsSoon(t, func() error {
+		if pc := testQueue.getProcessed(); pc != beforeProcessCount+1 {
+			return errors.Errorf("expected %d processed replicas; got %d", beforeProcessCount+1, pc)
+		}
+		if v := bq.successes.Count(); v != beforeSuccessCount+1 {
+			return errors.Errorf("expected %d processed replicas; got %d", beforeSuccessCount+1, v)
+		}
+		if v := bq.failures.Count(); v != beforeFailureCount {
+			return errors.Errorf("expected %d failed replicas; got %d", beforeFailureCount, v)
+		}
+		if v := bq.pending.Value(); v != 0 {
+			return errors.Errorf("expected 0 pending replicas; got %d", v)
+		}
+		if v := bq.purgatory.Value(); v != 0 {
+			return errors.Errorf("expected 0 purgatory replicas; got %d", v)
+		}
+		return nil
+	})
 }
 
 type processTimeoutQueueImpl struct {