rac2: fix decrement of kvflowcontrol.send_queue.scheduled.force_flush

Relates to cockroachlabs/support#3450

Epic: none

Release note (bug fix): The kvflowcontrol.send_queue.scheduled.force_flush
was missing a decrement, resulting in a value of > 0 even when there was
no ongoing force flush. This is fixed by this change.

Author: sumeerbhola

pkg/kv/kvserver/kvflowcontrol/rac2/range_controller.go

@@ -2682,7 +2682,7 @@ func (rs *replicaState) scheduledRaftMuLocked(
 		// knowledge will become known in the next
 		// rangeController.HandleRaftEventRaftMuLocked, which will happen at the
 		// next tick. We accept a latency hiccup in this case for now.
-		rss.mu.sendQueue.forceFlushStopIndex = 0
+		rss.setForceFlushStopIndexRaftMuAndStreamLocked(0)
 	}
 	forceFlushNeedsToPause := forceFlushActiveAndPaused()
 	watchForTokens :=
@@ -2768,7 +2768,7 @@ func (rss *replicaSendStream) handleReadyEntriesRaftMuAndStreamLocked(
 				rss.startForceFlushRaftMuAndStreamLocked(ctx, directive.forceFlushStopIndex)
 			} else {
 				if rss.mu.sendQueue.forceFlushStopIndex != directive.forceFlushStopIndex {
-					rss.mu.sendQueue.forceFlushStopIndex = directive.forceFlushStopIndex
+					rss.setForceFlushStopIndexRaftMuAndStreamLocked(directive.forceFlushStopIndex)
 				}
 				if wasExceedingInflightBytesThreshold &&
 					!rss.reachedInflightBytesThresholdRaftMuAndStreamLocked() {
@@ -2780,8 +2780,7 @@ func (rss *replicaSendStream) handleReadyEntriesRaftMuAndStreamLocked(
 			if rss.mu.sendQueue.forceFlushStopIndex.active() {
 				// Must have a send-queue, so sendingEntries should stay empty (these
 				// will be queued).
-				rss.mu.sendQueue.forceFlushStopIndex = 0
-				rss.parent.parent.opts.RangeControllerMetrics.SendQueue.ForceFlushedScheduledCount.Dec(1)
+				rss.setForceFlushStopIndexRaftMuAndStreamLocked(0)
 				rss.startAttemptingToEmptySendQueueViaWatcherStreamLocked(ctx)
 				if directive.hasSendTokens {
 					panic(errors.AssertionFailedf("hasSendTokens true despite send-queue"))
@@ -2998,8 +2997,7 @@ func (rss *replicaSendStream) startForceFlushRaftMuAndStreamLocked(
 ) {
 	rss.parent.parent.opts.ReplicaMutexAsserter.RaftMuAssertHeld()
 	rss.mu.AssertHeld()
-	rss.parent.parent.opts.RangeControllerMetrics.SendQueue.ForceFlushedScheduledCount.Inc(1)
-	rss.mu.sendQueue.forceFlushStopIndex = forceFlushStopIndex
+	rss.setForceFlushStopIndexRaftMuAndStreamLocked(forceFlushStopIndex)
 	if !rss.reachedInflightBytesThresholdRaftMuAndStreamLocked() {
 		rss.parent.parent.scheduleReplica(rss.parent.replicaID)
 	}
@@ -3136,8 +3134,7 @@ func (rss *replicaSendStream) stopAttemptingToEmptySendQueueRaftMuAndStreamLocke
 	rss.parent.parent.opts.ReplicaMutexAsserter.RaftMuAssertHeld()
 	rss.mu.AssertHeld()
 	if rss.mu.sendQueue.forceFlushStopIndex.active() {
-		rss.mu.sendQueue.forceFlushStopIndex = 0
-		rss.parent.parent.opts.RangeControllerMetrics.SendQueue.ForceFlushedScheduledCount.Dec(1)
+		rss.setForceFlushStopIndexRaftMuAndStreamLocked(0)
 	}
 	rss.stopAttemptingToEmptySendQueueViaWatcherRaftMuAndStreamLocked(ctx, disconnect)
 }
@@ -3167,6 +3164,21 @@ func (rss *replicaSendStream) stopAttemptingToEmptySendQueueViaWatcherRaftMuAndS
 	}
 }
 
+func (rss *replicaSendStream) setForceFlushStopIndexRaftMuAndStreamLocked(
+	index forceFlushStopIndex,
+) {
+	rss.parent.parent.opts.ReplicaMutexAsserter.RaftMuAssertHeld()
+	rss.mu.AssertHeld()
+	nextIsActive := index.active()
+	prevIsActive := rss.mu.sendQueue.forceFlushStopIndex.active()
+	if !prevIsActive && nextIsActive {
+		rss.parent.parent.opts.RangeControllerMetrics.SendQueue.ForceFlushedScheduledCount.Inc(1)
+	} else if prevIsActive && !nextIsActive {
+		rss.parent.parent.opts.RangeControllerMetrics.SendQueue.ForceFlushedScheduledCount.Dec(1)
+	}
+	rss.mu.sendQueue.forceFlushStopIndex = index
+}
+
 // Requires that send-queue is non-empty. Note that it is possible that all
 // the entries in the send-queue are not subject to replication admission
 // control, and we will still wait for non-zero tokens. This is considered