Merge #152596 #152792

152596: kvserver: requeue on priority inversion for replicate queue r=tbg a=wenyihu6

Part of: #152022
Release note: none

---


**kvserver: add PriorityInversionRequeue**

This commit adds a new cluster setting PriorityInversionRequeue that controls
whether the replicate queue should requeue replicas when their priority at
enqueue time differs significantly from their priority at processing time
(e.g. dropping from top 3 to the lowest priority).

---

**kvserver: requeue on priority inversion for replicate queue**

Previously, a replica could enter the queue with high priority but, by the time
it was processed, the action planned for this replica may have a low priority,
causing us to perform low priority work. Specifically, we are mostly worried
about cases where the priority changes from any of the repair actions to
consider rebalance. Rebalancing could take a long time and block other ranges
enqueued with actual repair action needed. This commit ensures that such
replicas are requeued instead, avoiding priority inversions.

---

**kvserver: use priorityInversionLogEveryN**

Previously, replicateQueue used V(2) to log info on priority inverted replicas
because I wanted visibility into every case without missing any replicas. On
reflection, the individual cases aren’t that interesting - it’s the overall
volume that matters, which we can track through metrics. This commit changes it
so that we just rate limit priority inversions every 3 seconds.

---

**kvserver: improve comments for PriorityInversionRequeue**

This commit improves the comments for PriorityInversionRequeue and clarifies the
contracts around action.Priority().

---

**allocator: small refactor for CheckPriorityInversion**

This commit refactors CheckPriorityInversion.

---

**allocator: add TestAllocatorPriorityInvariance**

This commit adds the TestAllocatorPriorityInvariance test, which acts as a
regression safeguard when new actions are added to AllocatorAction, ensuring the
contract is upheld. See action.Priority() and ComputeAction() for more details
on the contract.

---

**kvserver: guard inversion check and requeue behind PriorityInversionRequeue**

Previously, we introduced the PriorityInversionRequeue cluster setting, intended
for backport, to handle cases where a range was enqueued with a high-priority
repair action but, at processing time, a low-priority rebalance action was
computed. In such cases, the caller re-adds the range to the queue under its
updated priority. Although the cluster setting guards this requeue behavior, the
inversion check always ran unconditionally, reducing backport safety. This
commit updates the logic so that the cluster setting guard both the inversion
check and the requeue behavior.

---

**kvserver: move priority inversion check before applyChange**

Previously, we checked for priority inversion before planning errors, which
meant we could return requeue = true even when a planning error occurred. This
commit changes it so that planning errors should take higher precedence over a
priority inversion error. rq.processOneChange now returns early if there is a
planning error and only check for priority inversion right before applying a
change.

---

**kvserver: check for requeue before error checking in rq.process**

Previously, we checked for requeue right before returning for both nil and
non-nil errors, making the code harder to follow. This commit refactors
replicateQueue.process to requeue replicas before checking for errors.


152792: kvserver: add onProcessResult and onEnqueueResult to processCallback r=tbg a=wenyihu6

Part of: #151847
Epic: none 

---

**kvserver: use non-blocking send on errors for maybeBackpressureBatch**

maybeBackpressureBatch registers a callback with the split queue for replicas
that are too large relative to their split size. This backpressures the range to
stop it from growing and prevent new writes from blocking a pending split. The
callback is invoked when the split queue finishes processing the replica.

Previously, the error channel used in the callback had a size of 1 and performed
blocking sends. This was safe because the base queue only sent a single error,
and by the time maybeBackpressureBatch returned, the callback was guaranteed to
have been consumed, and no additional sends would occur.

Future commits will allow the callback to be invoked multiple times (although it
should only twice at most). To be safe and avoid potential deadlocks from
multiple sends after maybeBackpressureBatch already returns, this commit makes
the error send non-blocking. If the channel is already full, the error is
dropped, which is acceptable since we only care about observing the completion
of the replica processing at least once.

---

**kvserver: return baseQueueAsyncRateLimited from bq.Async**

baseQueue.Async may return immediately as a noop if the semaphore does not
available capacity and the wait parameter is false. Previously, this case
returned no error, leaving the caller unaware that the request was dropped. This
commit changes the behavior to return a baseQueueAsyncRateLimited error,
allowing callers to detect and handle the condition.

---

**kvserver: add onProcessResult and onEnqueueResult to processCallback**

The base queue already supports registering callbacks that are invoked with the
processing result of replicas once they are processed. However, replicas may
fail before reaching that stage (e.g., failing to enqueue or dropped early).
This commit extends the mechanism to also report enqueue results, allowing
callers to detect failures earlier. Currently, only
decommissioningNudger.maybeEnqueueProblemRange uses this.

Note that one behavior change is introduced: previously, a registered callback
would fire only once with the processing result and not again if the replica was
later processed by the purgatory queue. With this change, the callback may now
be invoked twice.

---

**kvserver: add TestBaseQueueCallback**

This commit adds TestBaseQueueCallbackOnEnqueueResult and
TestBaseQueueCallbackOnProcessResult to verify that callbacks are correctly
invoked with both enqueue and process results.


Co-authored-by: wenyihu6 <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/allocator/allocatorimpl/allocator.go

@@ -139,6 +139,7 @@ const (
 	AllocatorConsiderRebalance
 	AllocatorRangeUnavailable
 	AllocatorFinalizeAtomicReplicationChange
+	AllocatorMaxPriority
 )
 
 // Add indicates an action adding a replica.
@@ -252,10 +253,27 @@ func (a AllocatorAction) SafeValue() {}
 // range. Within a given range, the ordering of the various checks inside
 // `Allocator.computeAction` determines which repair/rebalancing actions are
 // taken before the others.
+//
+// NB: Priorities should be non-negative and should be spaced in multiples of
+// 100 unless you believe they should belong to the same priority category.
+// AllocatorNoop should have the lowest priority. CheckPriorityInversion depends
+// on this contract. In most cases, the allocator returns a priority that
+// matches the definitions below. For AllocatorAddVoter,
+// AllocatorRemoveDeadVoter, and AllocatorRemoveVoter, the priority may be
+// adjusted (see ComputeAction for details), but the adjustment is expected to
+// be small (<49).
+//
+// Exceptions: AllocatorFinalizeAtomicReplicationChange, AllocatorRemoveLearner,
+// and AllocatorReplaceDeadVoter violates the spacing of 100. These cases
+// predate this comment, so we allow them as they belong to the same general
+// priority category.
 func (a AllocatorAction) Priority() float64 {
+	const maxPriority = 12002
 	switch a {
+	case AllocatorMaxPriority:
+		return maxPriority
 	case AllocatorFinalizeAtomicReplicationChange:
-		return 12002
+		return maxPriority
 	case AllocatorRemoveLearner:
 		return 12001
 	case AllocatorReplaceDeadVoter:
@@ -975,6 +993,49 @@ func (a *Allocator) ComputeAction(
 	return action, priority
 }
 
+// computeAction determines the action to take on a range along with its
+// priority.
+//
+// NB: The returned priority may include a small adjustment and therefore might
+// not exactly match action.Priority(). See AllocatorAddVoter,
+// AllocatorRemoveDeadVoter, AllocatorRemoveVoter below. The adjustment should
+// be <49 with two assumptions below. New uses on this contract should be
+// avoided since the assumptions are not strong guarantees (especially the
+// second one).
+//
+// The claim that the adjustment is < 49 has two assumptions:
+// 1. min(num_replicas,total_nodes) in zone configuration is < 98.
+// 2. when ranges are not under-replicated, the difference between
+// min(num_replicas,total_nodes)/2-1 and existing_replicas is < 49.
+//
+// neededVoters <= min(num_replicas,total_nodes)
+// desiredQuorum = neededVoters/2-1
+// quorum = haveVoters/2-1
+//
+// For AllocatorAddVoter, we know haveVoters < neededVoters
+// adjustment = desiredQuorum-haveVoters = neededVoters/2-1-haveVoters
+// To find the worst case (largest adjustment),
+//  1. haveVoters = neededVoters-1,
+//     adjustment = neededVoters/2-1-(neededVoters-1)
+//     = neededVoters/2-neededVoters = -neededVoters/2
+//  2. haveVoters = 0
+//     adjustement = neededVoters/2-1
+//
+// In order for adjustment to be <49, neededVoters/2<49 => neededVoters<98.
+// Hence the first assumption.
+//
+// For AllocatorRemoveDeadVoter, we know haveVoters >= neededVoters
+// adjustment = desiredQuorum-haveVoters = neededVoters/2-1-haveVoters
+// To find the worst case (largest adjustment),
+// 1. neededVoters/2-1 is much larger than haveVoters: given haveVoters >=
+// neededVoters, haveVoters/2-1 >= neededVoters/2-1. So this case is impossible.
+// 2. neededVoters/2-1 is much smaller than haveVoters: since ranges could be
+// over-replicated, theoretically speaking, there may be no upper bounds on
+// haveVoters. In order for adjustment to be < 49, we can only make an
+// assumption here that the difference between neededVoters/2-1 and haveVoters
+// cannot be >= 49 in this case.
+//
+// For AllocatorRemoveVoter, adjustment is haveVoters%2 = 0 or 1 < 49.
 func (a *Allocator) computeAction(
 	ctx context.Context,
 	storePool storepool.AllocatorStorePool,
@@ -3245,3 +3306,65 @@ func replDescsToStoreIDs(descs []roachpb.ReplicaDescriptor) []roachpb.StoreID {
 	}
 	return ret
 }
+
+// roundToNearestPriorityCategory rounds a priority to the nearest 100. n should
+// be non-negative.
+func roundToNearestPriorityCategory(n float64) float64 {
+	return math.Round(n/100.0) * 100
+}
+
+// CheckPriorityInversion returns whether there was a priority inversion (and
+// the range should not be processed at this time, since doing so could starve
+// higher-priority items), and whether the caller should re-add the range to the
+// queue (presumably under its new priority). A priority inversion happens if
+// the priority at enqueue time is higher than the priority corresponding to the
+// action computed at processing time. Caller should re-add the range to the
+// queue if it has gone from a repair action to lowest priority
+// (AllocatorConsiderRebalance).
+//
+// Note: Changing from AllocatorRangeUnavailable/AllocatorNoop to
+// AllocatorConsiderRebalance is not treated as a priority inversion. Going from
+// a repair action to AllocatorRangeUnavailable/AllocatorNoop is considered a
+// priority inversion but shouldRequeue = false.
+//
+// INVARIANT: shouldRequeue => isInversion
+func CheckPriorityInversion(
+	priorityAtEnqueue float64, actionAtProcessing AllocatorAction,
+) (isInversion bool, shouldRequeue bool) {
+	// NB: priorityAtEnqueue is -1 for callers such as scatter, dry runs, and
+	// manual queue runs. Priority inversion does not apply to these calls.
+	if priorityAtEnqueue == -1 {
+		return false, false
+	}
+
+	// NB: we need to check for when priorityAtEnqueue falls within the range
+	// of the allocator actions because store.Enqueue might enqueue things with
+	// a very high priority (1e5). In those cases, we do not want to requeue
+	// these actions or count it as an inversion.
+	withinPriorityRange := func(priority float64) bool {
+		return AllocatorNoop.Priority() <= priority && priority <= AllocatorMaxPriority.Priority()
+	}
+	if !withinPriorityRange(priorityAtEnqueue) {
+		return false, false
+	}
+
+	if priorityAtEnqueue > AllocatorConsiderRebalance.Priority() && actionAtProcessing == AllocatorConsiderRebalance {
+		return true, true
+	}
+
+	// NB: Usually, the priority at enqueue time should correspond to
+	// action.Priority(). However, for AllocatorAddVoter,
+	// AllocatorRemoveDeadVoter, AllocatorRemoveVoter, the priority can be
+	// adjusted at enqueue time (See ComputeAction for more details). However, we
+	// expect the adjustment to be relatively small (<49). So we round the
+	// priority to the nearest 100 to compare against
+	// actionAtProcessing.Priority(). Without this rounding, we might treat going
+	// from 10000 to 999 as an inversion, but it was just due to the adjustment.
+	// Note that priorities at AllocatorFinalizeAtomicReplicationChange,
+	// AllocatorRemoveLearner, and AllocatorReplaceDeadVoter will be rounded to
+	// the same priority. They are so close to each other, so we don't really
+	// count it as an inversion among them.
+	normPriorityAtEnqueue := roundToNearestPriorityCategory(priorityAtEnqueue)
+	isInversion = normPriorityAtEnqueue > actionAtProcessing.Priority()
+	return isInversion, false
+}

pkg/kv/kvserver/allocator/allocatorimpl/allocator_test.go

@@ -9629,3 +9629,181 @@ func TestAllocatorRebalanceTargetVoterConstraintUnsatisfied(t *testing.T) {
 		})
 	}
 }
+
+// TestRoundToNearestPriorityCategory tests the RoundToNearestPriorityCategory
+// function.
+func TestRoundToNearestPriorityCategory(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	testCases := []struct {
+		name     string
+		input    float64
+		expected float64
+	}{
+		{
+			name:     "zero",
+			input:    0.0,
+			expected: 0.0,
+		},
+		{
+			name:     "exact multiple of 100",
+			input:    100.0,
+			expected: 100.0,
+		},
+		{
+			name:     "round down to nearest 100",
+			input:    149.0,
+			expected: 100.0,
+		},
+		{
+			name:     "round up to nearest 100",
+			input:    151.0,
+			expected: 200.0,
+		},
+		{
+			name:     "negative exact multiple of 100",
+			input:    -200.0,
+			expected: -200.0,
+		},
+		{
+			name:     "negative round down to nearest 100",
+			input:    -249.0,
+			expected: -200.0,
+		},
+		{
+			name:     "negative round up to nearest 100",
+			input:    -251.0,
+			expected: -300.0,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.expected, roundToNearestPriorityCategory(tc.input))
+		})
+	}
+}
+
+// TestCheckPriorityInversion tests the CheckPriorityInversion function.
+func TestCheckPriorityInversion(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	for action := AllocatorNoop; action <= AllocatorFinalizeAtomicReplicationChange; action++ {
+		t.Run(action.String(), func(t *testing.T) {
+			if action == AllocatorConsiderRebalance || action == AllocatorNoop || action == AllocatorRangeUnavailable {
+				inversion, requeue := CheckPriorityInversion(action.Priority(), AllocatorConsiderRebalance)
+				require.False(t, inversion)
+				require.False(t, requeue)
+			} else {
+				inversion, requeue := CheckPriorityInversion(action.Priority(), AllocatorConsiderRebalance)
+				require.True(t, inversion)
+				require.True(t, requeue)
+			}
+		})
+	}
+
+	testCases := []struct {
+		name               string
+		priorityAtEnqueue  float64
+		actionAtProcessing AllocatorAction
+		expectedInversion  bool
+		expectedRequeue    bool
+	}{
+		{
+			name:               "AllocatorNoop at processing is noop",
+			priorityAtEnqueue:  AllocatorFinalizeAtomicReplicationChange.Priority(),
+			actionAtProcessing: AllocatorNoop,
+			expectedInversion:  true,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "AllocatorRangeUnavailable at processing is noop",
+			priorityAtEnqueue:  AllocatorFinalizeAtomicReplicationChange.Priority(),
+			actionAtProcessing: AllocatorRangeUnavailable,
+			expectedInversion:  true,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "priority -1 bypasses",
+			priorityAtEnqueue:  -1,
+			actionAtProcessing: AllocatorConsiderRebalance,
+			expectedInversion:  false,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "priority increase",
+			priorityAtEnqueue:  0,
+			actionAtProcessing: AllocatorFinalizeAtomicReplicationChange,
+			expectedInversion:  false,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "above range priority(1e5)",
+			priorityAtEnqueue:  1e5,
+			actionAtProcessing: AllocatorConsiderRebalance,
+			expectedInversion:  false,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "below range priority at -10",
+			priorityAtEnqueue:  -10,
+			actionAtProcessing: -100,
+			expectedInversion:  false,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "inversion but small priority changes",
+			priorityAtEnqueue:  AllocatorFinalizeAtomicReplicationChange.Priority(),
+			actionAtProcessing: AllocatorReplaceDecommissioningNonVoter,
+			expectedInversion:  true,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "inversion but small priority changes",
+			priorityAtEnqueue:  AllocatorRemoveDeadVoter.Priority(),
+			actionAtProcessing: AllocatorAddNonVoter,
+			expectedInversion:  true,
+			expectedRequeue:    false,
+		},
+		{
+			name:               "inversion but small priority changes",
+			priorityAtEnqueue:  AllocatorConsiderRebalance.Priority(),
+			actionAtProcessing: AllocatorNoop,
+			expectedInversion:  false,
+			expectedRequeue:    false,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			inversion, requeue := CheckPriorityInversion(tc.priorityAtEnqueue, tc.actionAtProcessing)
+			require.Equal(t, tc.expectedInversion, inversion)
+			require.Equal(t, tc.expectedRequeue, requeue)
+		})
+	}
+}
+
+// TestAllocatorPriorityInvariance verifies that allocator priorities remain
+// spaced in multiples of 100. This prevents regressions against the contract
+// relied on by CheckPriorityInversion. For details, see the comment above
+// action.Priority().
+func TestAllocatorPriorityInvariance(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	exceptions := map[AllocatorAction]struct{}{
+		AllocatorFinalizeAtomicReplicationChange: {},
+		AllocatorRemoveLearner:                   {},
+		AllocatorReplaceDeadVoter:                {},
+	}
+	lowestPriority := AllocatorNoop.Priority()
+	for action := AllocatorNoop; action < AllocatorMaxPriority; action++ {
+		require.GreaterOrEqualf(t, action.Priority(), lowestPriority,
+			"priority %f is less than AllocatorNoop: likely violating contract",
+			action.Priority())
+		if _, ok := exceptions[action]; !ok {
+			require.Equalf(t, int(action.Priority())%100, 0,
+				"priority %f is not a multiple of 100: likely violating contract",
+				action.Priority())
+
+		}
+	}
+}

pkg/kv/kvserver/mvcc_gc_queue.go

@@ -878,7 +878,7 @@ func (mgcq *mvccGCQueue) scanReplicasForHiPriGCHints(
 				if !isLeaseHolder {
 					return true
 				}
-				added, _ := mgcq.addInternal(ctx, desc, replica.ReplicaID(), deleteRangePriority)
+				added, _ := mgcq.addInternal(ctx, desc, replica.ReplicaID(), deleteRangePriority, noopProcessCallback)
 				if added {
 					mgcq.store.metrics.GCEnqueueHighPriority.Inc(1)
 					foundReplicas++