Merge #148605

148605: kvcoord: fix bug in bufferSize accounting in txnWriteBuffer r=arulajmani a=stevendanna

For a lock that was never held, we were over-decrementing the buffer size on savepoint rollback. I've added a regression test, but this bug was also caught by KVNemesis.

Epic: none
Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -655,7 +655,8 @@ func (twb *txnWriteBuffer) rollbackToSavepointLocked(ctx context.Context, s save
 	toDelete := make([]*bufferedWrite, 0)
 	it := twb.buffer.MakeIter()
 	for it.First(); it.Valid(); it.Next() {
-		if held := it.Cur().rollbackLockInfo(s.seqNum); !held {
+		hadLockInfo := it.Cur().lki != nil
+		if held := it.Cur().rollbackLockInfo(s.seqNum); hadLockInfo && !held {
 			// If we aren't still held, update our buffer size.
 			twb.bufferSize -= lockKeyInfoSize
 		}

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -1820,6 +1820,47 @@ func TestTxnWriteBufferRollbackToSavepoint(t *testing.T) {
 	require.IsType(t, &kvpb.EndTxnResponse{}, br.Responses[0].GetInner())
 }
 
+// TestRollbackNeverHeldLock is a regression test for a bug around incorrect
+// accounting of the buffer size for completely unlocked writes that were rolled
+// back.
+func TestRollbackNeverHeldLock(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	twb, mockSender := makeMockTxnWriteBuffer(cluster.MakeClusterSettings())
+
+	txn := makeTxnProto()
+	txn.Sequence = 10
+	txn.Sequence++
+	sp := &savepoint{seqNum: txn.Sequence}
+	twb.createSavepointLocked(ctx, sp)
+
+	txn.Sequence++
+	ba := &kvpb.BatchRequest{Header: kvpb.Header{Txn: &txn}}
+	ba.Add(delArgs(roachpb.Key("a"), txn.Sequence))
+
+	br, pErr := twb.SendLocked(ctx, ba)
+	require.Nil(t, pErr)
+	require.NotNil(t, br)
+
+	twb.rollbackToSavepointLocked(ctx, *sp)
+
+	// Commit the transaction.
+	ba = &kvpb.BatchRequest{Header: kvpb.Header{Txn: &txn}}
+	ba.Add(&kvpb.EndTxnRequest{Commit: true})
+
+	mockSender.MockSend(func(ba *kvpb.BatchRequest) (*kvpb.BatchResponse, *kvpb.Error) {
+		require.Len(t, ba.Requests, 1)
+		br = ba.CreateReply()
+		br.Txn = ba.Txn
+		return br, nil
+	})
+
+	br, pErr = twb.SendLocked(ctx, ba)
+	require.Nil(t, pErr)
+	require.NotNil(t, br)
+}
+
 // TestTxnWriteBufferRollbackToSavepointMidTxn tests the savepoint rollback
 // logic in the presence of explicit savepoints.
 func TestTxnWriteBufferRollbackToSavepointMidTxn(t *testing.T) {