kvserver: enqueue decom ranges at an interval behind a setting

When `kv.enqueue_in_replicate_queue_on_problem.interval` is set to a
positive non-zero value, leaseholder replicas of ranges which have
decommissioning replicas will be enqueued into the replicate queue every
`kv.enqueue_in_replicate_queue_on_problem.interval` interval.

When `kv.enqueue_in_replicate_queue_on_problem.interval` is set to 0,
no enqueueing on decommissioning will take place, outside of the regular
replica scanner.

A recommended value for users enabling the enqueue (non-zero), is at
least 15 minutes e.g.,

```
SET CLUSTER SETTING
kv.enqueue_in_replicate_queue_on_problem.interval='900s'
```

Resolves: #130085
Informs: #130199
Release note: None

Author: kvoli

pkg/kv/kvserver/replica.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/abortspan"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/allocator/allocatorimpl"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/batcheval"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/gc"
@@ -844,6 +845,12 @@ type Replica struct {
 	// loadBasedSplitter keeps information about load-based splitting.
 	loadBasedSplitter split.Decider
 
+	// lastProblemRangeReplicateEnqueueTime is the last time this replica was
+	// eagerly enqueued into the replicate queue due to being underreplicated
+	// or having a decommissioning replica. This is used to throttle enqueue
+	// attempts.
+	lastProblemRangeReplicateEnqueueTime atomic.Value
+
 	// unreachablesMu contains a set of remote ReplicaIDs that are to be reported
 	// as unreachable on the next raft tick.
 	unreachablesMu struct {
@@ -2356,3 +2363,57 @@ func (r *Replica) ReadProtectedTimestampsForTesting(ctx context.Context) (err er
 	ts, err = r.readProtectedTimestampsRLocked(ctx)
 	return err
 }
+
+// maybeEnqueueProblemRange will enqueue the replica for processing into the
+// replicate queue iff:
+//
+//   - The replica is the holder of a valid lease.
+//   - EnqueueProblemRangeInReplicateQueueInterval is enabled (set to a
+//     non-zero value)
+//   - The last time the replica was enqueued is longer than
+//     EnqueueProblemRangeInReplicateQueueInterval.
+//
+// The replica is enqueued at a decommissioning priority. Note that by default,
+// this behavior is disabled (zero interval). Also note that this method should
+// NOT be called unless the range is known to require action e.g.,
+// decommissioning|underreplicated.
+//
+// NOTE: This method is motivated by a bug where decommissioning stalls because
+// a decommissioning range is not enqueued in the replicate queue in a timely
+// manner via the replica scanner, see #130199. This functionality is disabled
+// by default for this reason.
+func (r *Replica) maybeEnqueueProblemRange(
+	ctx context.Context, now time.Time, leaseValid, isLeaseholder bool,
+) {
+	// The method expects the caller to provide whether the lease is valid and
+	// the replica is the leaseholder for the range, so that it can avoid
+	// unnecessary work. We expect this method to be called in the context of
+	// updating metrics.
+	if !isLeaseholder || !leaseValid {
+		// The replicate queue will not process the replica without a valid lease.
+		// Nothing to do.
+		return
+	}
+
+	interval := EnqueueProblemRangeInReplicateQueueInterval.Get(&r.store.cfg.Settings.SV)
+	if interval == 0 {
+		// The setting is disabled.
+		return
+	}
+	lastTime := r.lastProblemRangeReplicateEnqueueTime.Load().(time.Time)
+	if lastTime.Add(interval).After(now) {
+		// The last time the replica was enqueued is less than the interval ago,
+		// nothing to do.
+		return
+	}
+	// The replica is the leaseholder for a range which requires action and it
+	// has been longer than EnqueueProblemRangeInReplicateQueueInterval since the
+	// last time it was enqueued. Try to swap the last time with now. We don't
+	// expect a race, however if the value changed underneath us we won't enqueue
+	// the replica as we lost the race.
+	if !r.lastProblemRangeReplicateEnqueueTime.CompareAndSwap(lastTime, now) {
+		return
+	}
+	r.store.replicateQueue.AddAsync(ctx, r,
+		allocatorimpl.AllocatorReplaceDecommissioningVoter.Priority())
+}

pkg/kv/kvserver/replica_init.go

@@ -128,6 +128,7 @@ func newUninitializedReplica(
 			store.rebalanceObjManager.Objective().ToSplitObjective(),
 		)
 	}
+	r.lastProblemRangeReplicateEnqueueTime.Store(store.Clock().PhysicalTime())
 
 	// NB: the state will be loaded when the replica gets initialized.
 	r.mu.state = uninitState

pkg/kv/kvserver/replicate_queue.go

@@ -91,6 +91,22 @@ var EnqueueInReplicateQueueOnSpanConfigUpdateEnabled = settings.RegisterBoolSett
 	false,
 )
 
+// EnqueueProblemRangeInReplicateQueueInterval controls the interval at which
+// problem ranges are enqueued into the replicate queue for processing, outside
+// of the normal scanner interval. A problem range is one which is
+// underreplicated or has a replica on a decommissioning store. The setting is
+// disabled when set to 0. By default, the setting is disabled.
+var EnqueueProblemRangeInReplicateQueueInterval = settings.RegisterDurationSetting(
+	settings.SystemOnly,
+	"kv.enqueue_in_replicate_queue_on_problem.interval",
+	"interval at which problem ranges are enqueued into the replicate queue for "+
+		"processing, outside of the normal scanner interval; a problem range is "+
+		"one which is underreplicated or has a replica on a decommissioning store, "+
+		"disabled when set to 0",
+	0,
+	settings.NonNegativeDuration,
+)
+
 var (
 	metaReplicateQueueAddReplicaCount = metric.Metadata{
 		Name:        "queue.replicate.addreplica",

pkg/kv/kvserver/replicate_queue_test.go

@@ -2778,3 +2778,69 @@ func TestReplicationChangesForRebalance(t *testing.T) {
 		})
 	}
 }
+
+// TestReplicateQueueDecommissionScannerDisabled asserts that decommissioning
+// replicas are replaced by the replicate queue despite the scanner being
+// disabled, when EnqueueProblemRangeInReplicateQueueInterval is set to a
+// non-zero value (enabled).
+func TestReplicateQueueDecommissionScannerDisabled(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// Enable enqueueing of problem ranges in the replicate queue at most once
+	// per second. We disable the scanner to ensure that the replicate queue
+	// doesn't rely on the scanner to process decommissioning replicas.
+	settings := cluster.MakeTestingClusterSettings()
+	kvserver.EnqueueProblemRangeInReplicateQueueInterval.Override(
+		context.Background(), &settings.SV, 1*time.Second)
+
+	ctx := context.Background()
+	tc := testcluster.StartTestCluster(t, 5, base.TestClusterArgs{
+		ReplicationMode: base.ReplicationManual,
+		ServerArgs: base.TestServerArgs{
+			Settings: settings,
+			// Disable the scanner.
+			ScanInterval:    100 * time.Hour,
+			ScanMinIdleTime: 100 * time.Hour,
+			ScanMaxIdleTime: 100 * time.Hour,
+		},
+	})
+	defer tc.Stopper().Stop(ctx)
+
+	decommissioningSrvIdx := 4
+	decommissioningSrv := tc.Server(decommissioningSrvIdx)
+	require.NoError(t, decommissioningSrv.Decommission(ctx,
+		livenesspb.MembershipStatus_DECOMMISSIONING,
+		[]roachpb.NodeID{tc.Server(decommissioningSrvIdx).NodeID()}))
+
+	// Ensure that the node is marked as decommissioning on every other node.
+	// Once this is set, we also know that the onDecommissioning callback has
+	// fired, which enqueues every range which is on the decommissioning node.
+	testutils.SucceedsSoon(t, func() error {
+		for i := 0; i < tc.NumServers(); i++ {
+			srv := tc.Server(i)
+			if _, exists := srv.DecommissioningNodeMap()[decommissioningSrv.NodeID()]; !exists {
+				return errors.Newf("node %d not detected to be decommissioning", decommissioningSrv.NodeID())
+			}
+		}
+		return nil
+	})
+
+	// Now add a replica to the decommissioning node and then enable the
+	// replicate queue. We expect that the replica will be removed after the
+	// decommissioning replica is noticed via maybeEnqueueProblemRange.
+	scratchKey := tc.ScratchRange(t)
+	tc.AddVotersOrFatal(t, scratchKey, tc.Target(decommissioningSrvIdx))
+	tc.ToggleReplicateQueues(true /* active */)
+	testutils.SucceedsSoon(t, func() error {
+		var descs []*roachpb.RangeDescriptor
+		tc.GetFirstStoreFromServer(t, decommissioningSrvIdx).VisitReplicas(func(r *kvserver.Replica) bool {
+			descs = append(descs, r.Desc())
+			return true
+		})
+		if len(descs) != 0 {
+			return errors.Errorf("expected no replicas, found %d: %v", len(descs), descs)
+		}
+		return nil
+	})
+}

pkg/kv/kvserver/store.go

@@ -3069,6 +3069,9 @@ func (s *Store) updateReplicationGauges(ctx context.Context) error {
 				overreplicatedRangeCount++
 			}
 			if metrics.Decommissioning {
+				// NB: Enqueue is disabled by default from here and throttled async if
+				// enabled.
+				rep.maybeEnqueueProblemRange(ctx, now.ToTimestamp().GoTime(), metrics.LeaseValid, metrics.Leaseholder)
 				decommissioningRangeCount++
 			}
 		}