kv: don't require consensus for no-op EndTxn commit/abort

Before this commit, EndTxn requests that hit the auto-GC path, did not resolve
any intents, and also did not find an existing transaction record would still
write a tombstone over their transaction record. This would force them to go
through Raft and miss the fast-path to skip consensus added in 6fcb3db. This
commit ensures that a tombstone is not written over a non-existing transaction
record, which allows this case to avoid Raft altogether.

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -202,11 +202,12 @@ func EndTxn(
 
 	// Fetch existing transaction.
 	var existingTxn roachpb.Transaction
-	if ok, err := storage.MVCCGetProto(
+	recordAlreadyExisted, err := storage.MVCCGetProto(
 		ctx, readWriter, key, hlc.Timestamp{}, &existingTxn, storage.MVCCGetOptions{},
-	); err != nil {
+	)
+	if err != nil {
 		return result.Result{}, err
-	} else if !ok {
+	} else if !recordAlreadyExisted {
 		// No existing transaction record was found - create one by writing it
 		// below in updateFinalizedTxn.
 		reply.Txn = h.Txn.Clone()
@@ -251,7 +252,7 @@ func EndTxn(
 					return result.Result{}, err
 				}
 				if err := updateFinalizedTxn(
-					ctx, readWriter, ms, key, args, reply.Txn, externalLocks,
+					ctx, readWriter, ms, key, args, reply.Txn, recordAlreadyExisted, externalLocks,
 				); err != nil {
 					return result.Result{}, err
 				}
@@ -330,7 +331,9 @@ func EndTxn(
 	if err != nil {
 		return result.Result{}, err
 	}
-	if err := updateFinalizedTxn(ctx, readWriter, ms, key, args, reply.Txn, externalLocks); err != nil {
+	if err := updateFinalizedTxn(
+		ctx, readWriter, ms, key, args, reply.Txn, recordAlreadyExisted, externalLocks,
+	); err != nil {
 		return result.Result{}, err
 	}
 
@@ -555,12 +558,19 @@ func updateFinalizedTxn(
 	key []byte,
 	args *roachpb.EndTxnRequest,
 	txn *roachpb.Transaction,
+	recordAlreadyExisted bool,
 	externalLocks []roachpb.Span,
 ) error {
 	if txnAutoGC && len(externalLocks) == 0 {
 		if log.V(2) {
 			log.Infof(ctx, "auto-gc'ed %s (%d locks)", txn.Short(), len(args.LockSpans))
 		}
+		if !recordAlreadyExisted {
+			// Nothing to delete, so there's no use writing a deletion tombstone. This
+			// can help avoid sending a proposal through Raft, if nothing else in the
+			// BatchRequest writes.
+			return nil
+		}
 		return storage.MVCCDelete(ctx, readWriter, ms, key, hlc.Timestamp{}, nil /* txn */)
 	}
 	txn.LockSpans = externalLocks

pkg/kv/kvserver/replica_test.go

@@ -9156,6 +9156,14 @@ func TestNoopRequestsNotProposed(t *testing.T) {
 	deleteReq := &roachpb.DeleteRequest{
 		RequestHeader: rh,
 	}
+	endTxnCommitReq := &roachpb.EndTxnRequest{
+		RequestHeader: rh,
+		Commit:        true,
+	}
+	endTxnAbortReq := &roachpb.EndTxnRequest{
+		RequestHeader: rh,
+		Commit:        true,
+	}
 	hbTxnReq := &roachpb.HeartbeatTxnRequest{
 		RequestHeader: rh,
 		Now:           cfg.Clock.Now(),
@@ -9238,6 +9246,42 @@ func TestNoopRequestsNotProposed(t *testing.T) {
 			// NB: a tombstone intent is written even if no value exists at the key.
 			expProposal: true,
 		},
+		{
+			name:        "end txn (commit) with auto-gc, without existing record",
+			useTxn:      true,
+			req:         endTxnCommitReq,
+			expProposal: false,
+		},
+		{
+			name:        "end txn (abort) with auto-gc, without existing record",
+			useTxn:      true,
+			req:         endTxnAbortReq,
+			expProposal: false,
+		},
+		{
+			name: "end txn (commit) with auto-gc, with existing record",
+			setup: func(ctx context.Context, repl *Replica) *roachpb.Error {
+				return sendReq(ctx, repl, hbTxnReq, txn)
+			},
+			useTxn:      true,
+			req:         endTxnCommitReq,
+			expProposal: true,
+		},
+		{
+			name: "end txn (abort) with auto-gc, with existing record",
+			setup: func(ctx context.Context, repl *Replica) *roachpb.Error {
+				return sendReq(ctx, repl, hbTxnReq, txn)
+			},
+			useTxn:      true,
+			req:         endTxnAbortReq,
+			expProposal: true,
+		},
+		{
+			name:        "heartbeat txn",
+			useTxn:      true,
+			req:         hbTxnReq,
+			expProposal: true,
+		},
 		{
 			name: "push txn req",
 			setup: func(ctx context.Context, repl *Replica) *roachpb.Error {