kvcoord: fix memory leak on rollbacks in txnWriteBuffer in some cases

Whenever we perform a rollback of the txnWriteBuffer, we need to remove
buffered values that are at or above the given sequence number. If the
whole buffered write is being rolled back, we delete it proper, but if
only some buffered values are rolled back, then we only remove them but
keep the `bufferedWrite`. Previously, we would decrement the buffer size
accordingly but forgot to also unset the `bufferedValue`s that could be
of non-trivial size. This is now fixed.

Release note: None

Author: yuzefovich

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -673,6 +673,12 @@ func (twb *txnWriteBuffer) rollbackToSavepointLocked(ctx context.Context, s save
 		// Update size bookkeeping for the values we're rolling back.
 		for i := idx; i < len(bufferedVals); i++ {
 			twb.bufferSize -= bufferedVals[i].size()
+			// Lose reference to the value since we're no longer tracking its
+			// memory footprint.
+			// TODO(yuzefovich): we also decremented the buffer size by the
+			// struct overhead, yet we keep reusing the same slice, so we have
+			// some slop in accounting.
+			bufferedVals[i] = bufferedValue{}
 		}
 		// Rollback writes by truncating the buffered values.
 		it.Cur().vals = bufferedVals[:idx]

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -1751,7 +1751,7 @@ func TestTxnWriteBufferRollbackToSavepoint(t *testing.T) {
 	require.Len(t, br.Responses, 2)
 	require.IsType(t, &kvpb.PutResponse{}, br.Responses[0].GetInner())
 	require.IsType(t, &kvpb.DeleteResponse{}, br.Responses[1].GetInner())
-	// Verify the
+	// Verify the state of the write buffer.
 	expBufferedWrites := []bufferedWrite{
 		makeBufferedWrite(keyA, makeBufferedValue("valA", 10)),
 		makeBufferedWrite(keyC, makeBufferedValue("", 11)),
@@ -1797,6 +1797,13 @@ func TestTxnWriteBufferRollbackToSavepoint(t *testing.T) {
 		makeBufferedWrite(keyA, makeBufferedValue("valA", 10)),
 	}
 	require.Equal(t, expBufferedWrites, twb.testingBufferedWritesAsSlice())
+	// Additionally ensure that we don't have a leak of rolled back writes.
+	{
+		bw := twb.testingBufferedWritesAsSlice()[0]
+		require.Greaterf(t, cap(bw.vals), 1, "should have kept the capacity of the slice")
+		vals := bw.vals[:cap(bw.vals)]
+		require.Equalf(t, bufferedValue{}, vals[1], "should have lost reference to the rolled back value")
+	}
 
 	// Commit the transaction.
 	ba = &kvpb.BatchRequest{}