kv: grab raftMu during no-op writes with local gossip triggers

Fixes #68011.

As of 9f8c019, it is now possible to have no-op writes that do not go through
Raft but do set one of the gossip triggers. These gossip triggers require the
raftMu to be held, so we were running into trouble when handling the local
eval results above Raft.

For instance, we see this case when a transaction sets the system config
trigger and then performs a delete range over an empty span before
committing. In this case, the transaction will have no intents to
remove, so it can auto-GC its record during an EndTxn. If its record was
never written in the first place, this is a no-op (as of 9f8c019).

There appear to be three ways we could solve this:
1. we can avoid setting gossip triggers on transactions that don't perform
   any writes.
2. we can force EndTxn requests with gossip triggers to go through Raft even
   if they are otherwise no-ops.
3. we can properly handle gossip triggers on the above Raft local eval result
   path.

This commit opts for the third option.

pkg/kv/kvserver/replica_application_state_machine.go

@@ -1206,7 +1206,7 @@ func (sm *replicaStateMachine) ApplySideEffects(
 	if cmd.IsLocal() {
 		// Handle the LocalResult.
 		if cmd.localResult != nil {
-			sm.r.handleReadWriteLocalEvalResult(ctx, *cmd.localResult)
+			sm.r.handleReadWriteLocalEvalResult(ctx, *cmd.localResult, true /* withRaftMu */)
 		}
 
 		rejected := cmd.Rejected()

pkg/kv/kvserver/replica_proposal.go

@@ -641,7 +641,9 @@ func addSSTablePreApply(
 	return copied
 }
 
-func (r *Replica) handleReadWriteLocalEvalResult(ctx context.Context, lResult result.LocalResult) {
+func (r *Replica) handleReadWriteLocalEvalResult(
+	ctx context.Context, lResult result.LocalResult, withRaftMu bool,
+) {
 	// Fields for which no action is taken in this method are zeroed so that
 	// they don't trigger an assertion at the end of the method (which checks
 	// that all fields were handled).
@@ -707,21 +709,35 @@ func (r *Replica) handleReadWriteLocalEvalResult(ctx context.Context, lResult re
 		lResult.MaybeAddToSplitQueue = false
 	}
 
+	// The following three triggers require the raftMu to be held. If a
+	// trigger is present, acquire the mutex if it is not held already.
+	maybeAcquireRaftMu := func() func() {
+		if withRaftMu {
+			return func() {}
+		}
+		withRaftMu = true
+		r.raftMu.Lock()
+		return r.raftMu.Unlock
+	}
+
 	if lResult.MaybeGossipSystemConfig {
+		defer maybeAcquireRaftMu()()
 		if err := r.MaybeGossipSystemConfigRaftMuLocked(ctx); err != nil {
 			log.Errorf(ctx, "%v", err)
 		}
 		lResult.MaybeGossipSystemConfig = false
 	}
 
 	if lResult.MaybeGossipSystemConfigIfHaveFailure {
+		defer maybeAcquireRaftMu()()
 		if err := r.MaybeGossipSystemConfigIfHaveFailureRaftMuLocked(ctx); err != nil {
 			log.Errorf(ctx, "%v", err)
 		}
 		lResult.MaybeGossipSystemConfigIfHaveFailure = false
 	}
 
 	if lResult.MaybeGossipNodeLiveness != nil {
+		defer maybeAcquireRaftMu()()
 		if err := r.MaybeGossipNodeLivenessRaftMuLocked(ctx, *lResult.MaybeGossipNodeLiveness); err != nil {
 			log.Errorf(ctx, "%v", err)
 		}

pkg/kv/kvserver/replica_raft.go

@@ -115,7 +115,7 @@ func (r *Replica) evalAndPropose(
 	if proposal.command == nil {
 		intents := proposal.Local.DetachEncounteredIntents()
 		endTxns := proposal.Local.DetachEndTxns(pErr != nil /* alwaysOnly */)
-		r.handleReadWriteLocalEvalResult(ctx, *proposal.Local)
+		r.handleReadWriteLocalEvalResult(ctx, *proposal.Local, false /* withRaftMu */)
 
 		pr := proposalResult{
 			Reply:              proposal.Local.Reply,