kvcoord: always check for buffered writes when checking for locks

When reading through code for an unrelated reason we noted that there
was a call checking for acquired locks but not buffered writes. This is
concerning since the reasoning for why we need to check for acquired
locks almost always implies we should also check for buffered writes.

In the particular case of RollbackToSavepoint, I believe the omission
has no impact. Namely:

  - Previous writes have been cleared from the buffer and thus aren't in
    danger of failing to be rolled back.

  - Subsequent writes will bump the sequence number and thus those
    writes are not in danger of being erroneously rolled back.

  - Subsequent reads cannot read the data that should have been rolled
    back because it will have been cleared from the buffer.

Further, for real-world SQL transactions, we never have a buffered write
without having at least 1 lock.

However, for the avoidance fo doubt, we now check both and add a new
method to avoid missing checks in the future.

Fixes cockroachdb#155975
Release note: None

Author: stevendanna

pkg/kv/kvclient/kvcoord/txn_coord_sender.go

@@ -527,8 +527,7 @@ func (tc *TxnCoordSender) Send(
 		return nil, pErr
 	}
 
-	if ba.IsSingleEndTxnRequest() && !tc.interceptorAlloc.txnPipeliner.hasAcquiredLocks() &&
-		!tc.interceptorAlloc.txnWriteBuffer.hasBufferedWrites() {
+	if ba.IsSingleEndTxnRequest() && !tc.hasAcquiredLocksOrBufferedWritesLocked() {
 		return nil, tc.finalizeNonLockingTxnLocked(ctx, ba)
 	}
 
@@ -1724,3 +1723,7 @@ func (tc *TxnCoordSender) hasPerformedWritesLocked() bool {
 func (tc *TxnCoordSender) hasBufferedWritesLocked() bool {
 	return tc.interceptorAlloc.txnWriteBuffer.hasBufferedWrites()
 }
+
+func (tc *TxnCoordSender) hasAcquiredLocksOrBufferedWritesLocked() bool {
+	return tc.interceptorAlloc.txnPipeliner.hasAcquiredLocks() || tc.interceptorAlloc.txnWriteBuffer.hasBufferedWrites()
+}

pkg/kv/kvclient/kvcoord/txn_coord_sender_savepoints.go

@@ -89,12 +89,12 @@ func (tc *TxnCoordSender) CreateSavepoint(ctx context.Context) (kv.SavepointToke
 	// We also increment the write sequence if any writes are buffered since
 	// a buffered write will not appears as a lock acquisition in the
 	// pipeliner. For SQL-generated transactions, this additional check is
-	// unnecessary since we won't ever have buffered writes wihout also at
+	// unnecessary since we won't ever have buffered writes without also at
 	// least 1 locked row. However, for other types of KV transactions such
 	// as those generated by KVNemesis we may have blind Put or Delete
 	// requests that would be erroneously rolled back if we didn't increment
 	// the sequence here.
-	if tc.interceptorAlloc.txnPipeliner.hasAcquiredLocks() || tc.interceptorAlloc.txnWriteBuffer.hasBufferedWrites() {
+	if tc.hasAcquiredLocksOrBufferedWritesLocked() {
 		if err := tc.interceptorAlloc.txnSeqNumAllocator.stepWriteSeqLocked(ctx); err != nil {
 			return nil, err
 		}
@@ -160,7 +160,7 @@ func (tc *TxnCoordSender) RollbackToSavepoint(ctx context.Context, s kv.Savepoin
 	// TODO(nvanbenschoten): once #113765 is resolved, we should make this
 	// unconditional and push the write sequence increment into
 	// txnSeqNumAllocator.rollbackToSavepointLocked.
-	if tc.interceptorAlloc.txnPipeliner.hasAcquiredLocks() {
+	if tc.hasAcquiredLocksOrBufferedWritesLocked() {
 		tc.mu.txn.AddIgnoredSeqNumRange(
 			enginepb.IgnoredSeqNumRange{
 				Start: sp.seqNum, End: tc.interceptorAlloc.txnSeqNumAllocator.writeSeq,