kvclient: de-flake tests due to metamorphic max_buffer_size

This commit adjusts the relevant tests that can produce a different
output if the txnWriteBuffer size is set too low (by ensuring a lower
bound on the cluster setting). All affected tests need at most 2KiB
which is the value I decided to use everywhere for consistency.

Additionally, in order to improve the test coverage, it reduces the
allowed range of metamorphic values to be in 1B-10KiB range.

`buffered_writes_lock_loss` test seemingly deadlocks with 1B buffer
size, so a separate issue is filed to investigate that.

Release note: None

Author: yuzefovich

pkg/ccl/logictestccl/testdata/logic_test/buffered_writes_lock_loss

@@ -1,6 +1,10 @@
 # LogicTest: !3node-tenant !local-mixed-25.2
 # cluster-opt: disable-mvcc-range-tombstones-for-point-deletes
 
+# TODO(#151895): investigate why setting it to 1B deadlocks the test.
+statement ok
+SET CLUSTER SETTING kv.transaction.write_buffering.max_buffer_size = '2KiB';
+
 statement ok
 SET CLUSTER SETTING sql.txn.write_buffering_for_weak_isolation.enabled=true
 

pkg/ccl/logictestccl/testdata/logic_test/cluster_locks_tenant_write_buffering

@@ -3,6 +3,9 @@
 statement ok
 SET kv_transaction_buffered_writes_enabled = true;
 
+statement ok
+SET CLUSTER SETTING kv.transaction.write_buffering.max_buffer_size = '2KiB';
+
 # Create a table, write a row, lock it, then switch users.
 statement ok
 CREATE TABLE t (k STRING PRIMARY KEY, v STRING, FAMILY (k,v))

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -59,15 +59,18 @@ var bufferedWritesGetTransformEnabled = settings.RegisterBoolSetting(
 )
 
 const defaultBufferSize = 1 << 22 // 4MB
-var bufferedWritesMaxBufferSize = settings.RegisterByteSizeSetting(
+
+// BufferedWritesMaxBufferSize controls the per-txn buffer size. It's exported
+// only to be accessed from tests.
+var BufferedWritesMaxBufferSize = settings.RegisterByteSizeSetting(
 	settings.ApplicationLevel,
 	"kv.transaction.write_buffering.max_buffer_size",
 	"if non-zero, defines that maximum size of the "+
 		"buffer that will be used to buffer transactional writes per-transaction",
 	int64(metamorphic.ConstantWithTestRange("kv.transaction.write_buffering.max_buffer_size",
 		defaultBufferSize, // default
 		1,                 // min
-		defaultBufferSize, // max
+		10<<10,            // max, 10KiB
 	)),
 	settings.NonNegativeInt,
 	settings.WithPublic,
@@ -298,10 +301,10 @@ func (twb *txnWriteBuffer) SendLocked(
 	// Check if buffering writes from the supplied batch will run us over
 	// budget. If it will, we shouldn't buffer writes from the current batch,
 	// and flush the buffer.
-	maxSize := bufferedWritesMaxBufferSize.Get(&twb.st.SV)
+	maxSize := BufferedWritesMaxBufferSize.Get(&twb.st.SV)
 	bufSize := twb.estimateSize(ba, cfg) + twb.bufferSize
 
-	// NB: if bufferedWritesMaxBufferSize is set to 0 then we effectively disable
+	// NB: if BufferedWritesMaxBufferSize is set to 0 then we effectively disable
 	// any buffer limiting.
 	if maxSize != 0 && bufSize > maxSize {
 		twb.txnMetrics.TxnWriteBufferMemoryLimitExceeded.Inc(1)
@@ -1051,9 +1054,9 @@ func (twb *txnWriteBuffer) maybeMutateBatchMaxSpanRequestKeys(
 	}
 
 	// If the user has disabled a maximum buffer size, respect that.
-	maxSize := bufferedWritesMaxBufferSize.Get(&twb.st.SV)
+	maxSize := BufferedWritesMaxBufferSize.Get(&twb.st.SV)
 	if maxSize == 0 {
-		log.VEventf(ctx, 2, "allowing unbounded transformed locking scan because %s=0", bufferedWritesMaxBufferSize.Name())
+		log.VEventf(ctx, 2, "allowing unbounded transformed locking scan because %s=0", BufferedWritesMaxBufferSize.Name())
 		return
 	}
 

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_client_test.go

@@ -60,6 +60,7 @@ func TestTxnCoordSenderWriteBufferingDisablesPipelining(t *testing.T) {
 	require.NoError(t, db.Put(ctx, "test-key-a", "hello"))
 
 	bufferedWritesScanTransformEnabled.Override(ctx, &st.SV, false)
+	BufferedWritesMaxBufferSize.Override(ctx, &st.SV, defaultBufferSize)
 
 	// Without write buffering
 	require.NoError(t, db.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -32,7 +32,7 @@ func makeMockTxnWriteBuffer(
 	st := cluster.MakeClusterSettings()
 	bufferedWritesScanTransformEnabled.Override(ctx, &st.SV, true)
 	bufferedWritesGetTransformEnabled.Override(ctx, &st.SV, true)
-	bufferedWritesMaxBufferSize.Override(ctx, &st.SV, defaultBufferSize)
+	BufferedWritesMaxBufferSize.Override(ctx, &st.SV, defaultBufferSize)
 
 	var metrics TxnMetrics
 	if len(optionalMetrics) > 0 {
@@ -1586,7 +1586,7 @@ func TestTxnWriteBufferFlushesWhenOverBudget(t *testing.T) {
 
 	putAEstimate := int64(len(keyA)+valA.Size()) + bufferedWriteStructOverhead + bufferedValueStructOverhead
 
-	bufferedWritesMaxBufferSize.Override(ctx, &st.SV, putAEstimate)
+	BufferedWritesMaxBufferSize.Override(ctx, &st.SV, putAEstimate)
 
 	ba := &kvpb.BatchRequest{}
 	ba.Header = kvpb.Header{Txn: &txn}
@@ -1739,7 +1739,7 @@ func TestTxnWriteBufferLimitsSizeOfScans(t *testing.T) {
 				txn.Sequence = 10
 
 				bufferedWritesScanTransformEnabled.Override(ctx, &st.SV, true)
-				bufferedWritesMaxBufferSize.Override(ctx, &st.SV, tc.bufferSize)
+				BufferedWritesMaxBufferSize.Override(ctx, &st.SV, tc.bufferSize)
 
 				ba := &kvpb.BatchRequest{Header: kvpb.Header{Txn: &txn}}
 				if isReverse {
@@ -1777,7 +1777,7 @@ func TestTxnWriteBufferLimitsSizeOfScans(t *testing.T) {
 		txn := makeTxnProto()
 		txn.Sequence = 10
 
-		bufferedWritesMaxBufferSize.Override(ctx, &st.SV, 10*(lockKeyInfoSize+int64(len(keyA))))
+		BufferedWritesMaxBufferSize.Override(ctx, &st.SV, 10*(lockKeyInfoSize+int64(len(keyA))))
 
 		ba := &kvpb.BatchRequest{Header: kvpb.Header{Txn: &txn}}
 		ba.Add(&kvpb.ScanRequest{
@@ -2132,7 +2132,7 @@ func TestTxnWriteBufferSplitsBatchesWithSkipLocked(t *testing.T) {
 		br.Txn = ba.Txn
 		return br, nil
 	})
-	bufferedWritesMaxBufferSize.Override(ctx, &st.SV, 1)
+	BufferedWritesMaxBufferSize.Override(ctx, &st.SV, 1)
 	ba = &kvpb.BatchRequest{
 		Header: kvpb.Header{
 			Txn:        &txn,
@@ -2533,7 +2533,7 @@ func TestTxnWriteBufferHasBufferedAllPrecedingWrites(t *testing.T) {
 		{
 			name: "flushed due to size limit",
 			setup: func(twb *txnWriteBuffer) {
-				bufferedWritesMaxBufferSize.Override(context.Background(), &twb.st.SV, 1)
+				BufferedWritesMaxBufferSize.Override(context.Background(), &twb.st.SV, 1)
 			},
 			ba: func(ba *kvpb.BatchRequest) {
 				putA := putArgs(keyA, "valA", txn.Sequence)

pkg/kv/kvclient/kvcoord/txn_test.go

@@ -1057,6 +1057,7 @@ func TestTxnContinueAfterCputError(t *testing.T) {
 		ctx := context.Background()
 		s := createTestDB(t)
 		defer s.Stop()
+		kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 		txn := s.DB.NewTxn(ctx, "test txn")
 
@@ -1097,6 +1098,7 @@ func TestTxnDeleteResponse(t *testing.T) {
 		ctx := context.Background()
 		s := createTestDB(t)
 		defer s.Stop()
+		kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 		txn := s.DB.NewTxn(ctx, "test txn")
 
@@ -1610,6 +1612,7 @@ func TestTxnBasicBufferedWrites(t *testing.T) {
 	defer log.Scope(t).Close(t)
 	s := createTestDB(t)
 	defer s.Stop()
+	kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 	testutils.RunTrueAndFalse(t, "commit", func(t *testing.T, commit bool) {
 		ctx := context.Background()
@@ -1791,6 +1794,7 @@ func TestTxnBufferedWritesOverlappingScan(t *testing.T) {
 	defer log.Scope(t).Close(t)
 	s := createTestDB(t)
 	defer s.Stop()
+	kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 	extractKVs := func(rows []roachpb.KeyValue, batchResponses [][]byte) []roachpb.KeyValue {
 		if rows != nil {
@@ -2031,6 +2035,7 @@ func TestTxnBufferedWritesConditionalPuts(t *testing.T) {
 	defer log.Scope(t).Close(t)
 	s := createTestDB(t)
 	defer s.Stop()
+	kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 	testutils.RunTrueAndFalse(t, "commit", func(t *testing.T, commit bool) {
 		ctx := context.Background()
@@ -2137,6 +2142,7 @@ func TestTxnBufferedWritesRollbackToSavepointAllBuffered(t *testing.T) {
 	defer log.Scope(t).Close(t)
 	s := createTestDB(t)
 	defer s.Stop()
+	kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 	ctx := context.Background()
 	value1 := []byte("value1")
@@ -2196,6 +2202,7 @@ func TestTxnBufferedWriteRetriesCorrectly(t *testing.T) {
 		TestingRequestFilter: reqFilter,
 	})
 	defer s.Stop()
+	kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 	rng, _ := randutil.NewTestRand()
 
@@ -2233,6 +2240,7 @@ func TestTxnBufferedWriteReadYourOwnWrites(t *testing.T) {
 
 	s := createTestDB(t)
 	defer s.Stop()
+	kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 	value1 := []byte("value1")
 	value21 := []byte("value21")
@@ -2412,6 +2420,7 @@ func TestTxnBufferedWritesOmitAbortSpanChecks(t *testing.T) {
 		},
 	})
 	defer s.Stop()
+	kvcoord.BufferedWritesMaxBufferSize.Override(context.Background(), s.DB.SettingsValues(), 2<<10 /* 2KiB */)
 
 	value1 := []byte("value1")
 	valueConflict := []byte("conflict")

pkg/sql/logictest/testdata/logic_test/as_of

@@ -154,6 +154,10 @@ ROLLBACK
 statement ok
 SET kv_transaction_buffered_writes_enabled = true;
 
+# Ensure that the write below is buffered and not flushed.
+statement ok
+SET CLUSTER SETTING kv.transaction.write_buffering.max_buffer_size = '2KiB';
+
 statement ok
 BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE
 

pkg/sql/logictest/testdata/logic_test/buffered_writes

@@ -3,6 +3,13 @@
 statement ok
 SET kv_transaction_buffered_writes_enabled=true
 
+# The buffer size can be changed metamoprhically, and if it happens to be too
+# small (around 500B), then some of the txns below might flush the buffer,
+# resulting in different KV requests. To ensure deterministic traces we always
+# update the buffer size to be large enough for txns in this file.
+statement ok
+SET CLUSTER SETTING kv.transaction.write_buffering.max_buffer_size = '2KiB';
+
 statement ok
 CREATE TABLE t1 (pk int primary key, v int, FAMILY (pk, v))
 

pkg/sql/logictest/testdata/logic_test/cluster_locks_write_buffering

@@ -3,6 +3,9 @@
 statement ok
 SET kv_transaction_buffered_writes_enabled = true;
 
+statement ok
+SET CLUSTER SETTING kv.transaction.write_buffering.max_buffer_size = '2KiB';
+
 # This test uses local-read-committed so that it can also test locking behavior
 # with READ COMMITTED transactions. However, we'll use a default of SERIALIZABLE
 # for all transactions.

pkg/sql/opt/exec/execbuilder/testdata/autocommit

@@ -10,6 +10,13 @@
 statement ok
 SET kv_transaction_buffered_writes_enabled = true
 
+# The buffer size can be changed metamoprhically, and if it happens to be too
+# small (around 500B), then some of the txns below might flush the buffer,
+# resulting in different KV requests. To ensure deterministic traces we always
+# update the buffer size to be large enough for txns in this file.
+statement ok
+SET CLUSTER SETTING kv.transaction.write_buffering.max_buffer_size = '2KiB';
+
 statement ok
 CREATE TABLE ab (a INT PRIMARY KEY, b INT, FAMILY f1 (a, b))