Merge #148708

148708: kvcoord: fix memory leak on rollbacks in txnWriteBuffer in some cases r=yuzefovich a=yuzefovich

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.

Epic: None
Release note: None

Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: craig[bot]

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{}