mmaprototype: use LeaseDisposition instead of VoterIsLagging bool

We'll want to exercise this all properly in tests (for now it is
hooked up in production, but nowhere else). This will be done as
part of testing work for #156776.

Epic: CRDB-55052

Author: tbg

pkg/kv/kvserver/allocator/mmaprototype/allocator_state.go

@@ -486,10 +486,10 @@ func (a *allocatorState) rebalanceStores(
 				}
 				var candsSet candidateSet
 				for _, cand := range cands {
-					if a.cs.stores[cand.storeID].adjusted.replicas[rangeID].VoterIsLagging {
+					if disp := a.cs.stores[cand.storeID].adjusted.replicas[rangeID].LeaseDisposition; disp != LeaseDispositionOK {
 						// Don't transfer lease to a store that is lagging.
-						log.KvDistribution.Infof(ctx, "skipping store s%d for lease transfer: replica is lagging",
-							cand.storeID)
+						log.KvDistribution.Infof(ctx, "skipping store s%d for lease transfer: lease disposition %v",
+							cand.storeID, disp)
 						continue
 					}
 					candSls := a.cs.computeLoadSummary(ctx, cand.storeID, &means.storeLoad, &means.nodeLoad)

pkg/kv/kvserver/allocator/mmaprototype/cluster_state.go

@@ -117,18 +117,9 @@ func (rit ReplicaIDAndType) subsumesChange(prev, next ReplicaIDAndType) bool {
 
 type ReplicaState struct {
 	ReplicaIDAndType
-	// VoterIsLagging can be set for a VOTER_FULL replica that has fallen
-	// behind, i.e., it's matching log is less than the current committed log.
-	// It is a hint to the allocator not to transfer the lease to this replica.
-	VoterIsLagging bool
-	// TODO(kvoli,sumeerbhola): Consider adding in rac2.SendQueue information to
-	// prevent lease transfers to replicas which are not able to take the lease
-	// due to a send queue. Do we even need this, given the VoterIsLagging
-	// should be true whenever there is a send queue? I suppose if we are
-	// force-flushing to a replica, it is possible for VoterIsLagging to be
-	// false and the send queue to be non-empty. But we could incorporate the
-	// send-queue state into VoterIsLagging -- having two bools doesn't seem
-	// beneficial.
+	// LeaseDisposition can be set for a VOTER_FULL replica and communicates the
+	// availability of this replica for lease transfers.
+	LeaseDisposition LeaseDisposition
 }
 
 // ChangeID is a unique ID, in the context of this data-structure and when
@@ -820,7 +811,7 @@ func (s StoreIDAndReplicaState) String() string {
 
 // SafeFormat implements the redact.SafeFormatter interface.
 func (s StoreIDAndReplicaState) SafeFormat(w redact.SafePrinter, _ rune) {
-	w.Printf("s%v:%v voterIsLagging:%v", s.StoreID, s.ReplicaState.ReplicaIDAndType, s.ReplicaState.VoterIsLagging)
+	w.Printf("s%v:%v lease disposition:%v", s.StoreID, s.ReplicaState.ReplicaIDAndType, s.ReplicaState.LeaseDisposition)
 }
 
 // rangeState is periodically updated based on reporting by the leaseholder.

pkg/kv/kvserver/mma_replica_store.go

@@ -125,10 +125,17 @@ func constructRangeMsgReplicas(
 	// should pass scratch memory to avoid unnecessary allocation.
 	replicas := make([]mmaprototype.StoreIDAndReplicaState, 0, len(desc.InternalReplicas))
 	for _, repl := range desc.InternalReplicas {
+		leaseDisp := mmaprototype.LeaseDispositionOK
+		if repl.IsAnyVoter() && raftutil.ReplicaIsBehind(raftStatus, repl.ReplicaID) {
+			// TODO: a follower for which we have developed a send queue will likely
+			// be behind, but we could also explicitly use the send queue state to
+			// inform the disposition below.
+			leaseDisp = mmaprototype.LeaseDispositionRefusing
+		}
 		replica := mmaprototype.StoreIDAndReplicaState{
 			StoreID: repl.StoreID,
 			ReplicaState: mmaprototype.ReplicaState{
-				VoterIsLagging: repl.IsAnyVoter() && raftutil.ReplicaIsBehind(raftStatus, repl.ReplicaID),
+				LeaseDisposition: leaseDisp,
 				ReplicaIDAndType: mmaprototype.ReplicaIDAndType{
 					ReplicaID: repl.ReplicaID,
 					ReplicaType: mmaprototype.ReplicaType{