kvclient: add tests for CPut evaluation with buffered writes

ConditionFailedErrors returned by CPuts are special, in that they're
non-terminal. That means that a transaction is allowed to commit even
if it gets one of these.

This patch reverts changes to tests from da7a4eb, which assumed we won't
try to commit transactions that encountered ConditionFailedErrors. While here,
we also extend TestTxnContinueAfterCputError to run with buffered writes and
validate the transaction's effects upon commit.

All these tests fail as of this commit.

Epic: none

Release note: None

Author: arulajmani

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -1118,11 +1118,11 @@ func TestTxnWriteBufferDecomposesConditionalPuts(t *testing.T) {
 			require.IsType(t, &kvpb.ConditionFailedError{}, pErr.GoError())
 		}
 
-		// Lastly, commit or rollback the transaction. A put should only be
-		// flushed if the condition evaluated successfully.
+		// Lastly, commit the transaction. A put should only be flushed if the
+		// condition evaluated successfully.
 		ba = &kvpb.BatchRequest{}
 		ba.Header = kvpb.Header{Txn: &txn}
-		ba.Add(&kvpb.EndTxnRequest{Commit: condEvalSuccessful})
+		ba.Add(&kvpb.EndTxnRequest{Commit: true})
 
 		mockSender.MockSend(func(ba *kvpb.BatchRequest) (*kvpb.BatchResponse, *kvpb.Error) {
 			if condEvalSuccessful {

pkg/kv/kvclient/kvcoord/txn_test.go

@@ -1037,19 +1037,38 @@ func TestTxnCommitTimestampAdvancedByRefresh(t *testing.T) {
 func TestTxnContinueAfterCputError(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	ctx := context.Background()
-	s := createTestDB(t)
-	defer s.Stop()
 
-	txn := s.DB.NewTxn(ctx, "test txn")
-	// Note: Since we're expecting the CPut to fail, the massaging done by
-	// StrToCPutExistingValue() is not actually necessary.
-	expVal := kvclientutils.StrToCPutExistingValue("dummy")
-	err := txn.CPut(ctx, "a", "val", expVal)
-	require.IsType(t, &kvpb.ConditionFailedError{}, err)
+	testutils.RunTrueAndFalse(t, "bufferedWritesEnabled", func(t *testing.T, bufferedWritesEnabled bool) {
+		ctx := context.Background()
+		s := createTestDB(t)
+		defer s.Stop()
 
-	require.NoError(t, txn.Put(ctx, "a", "b'"))
-	require.NoError(t, txn.Commit(ctx))
+		txn := s.DB.NewTxn(ctx, "test txn")
+
+		if bufferedWritesEnabled {
+			txn.SetBufferedWritesEnabled(true)
+		}
+
+		// Note: Since we're expecting the CPut to fail, the massaging done by
+		// StrToCPutExistingValue() is not actually necessary.
+		expVal := kvclientutils.StrToCPutExistingValue("dummy")
+		err := txn.CPut(ctx, "a", "val", expVal)
+		require.IsType(t, &kvpb.ConditionFailedError{}, err)
+
+		// Write to a different key.
+		require.NoError(t, txn.Put(ctx, "b", "b'"))
+		require.NoError(t, txn.Commit(ctx))
+
+		// We've successfully commited the transaction. The write to key "b"
+		// should be visible whereas the write to keyA shouldn't.
+		val, err := s.DB.Get(ctx, "a")
+		require.NoError(t, err)
+		require.False(t, val.Exists())
+
+		val, err = s.DB.Get(ctx, "b")
+		require.NoError(t, err)
+		require.Equal(t, "b'", string(val.ValueBytes()))
+	})
 }
 
 // Test that a transaction can be used after a locking request returns a
@@ -2013,7 +2032,6 @@ func TestTxnBufferedWritesConditionalPuts(t *testing.T) {
 				if expErr {
 					require.Error(t, err)
 					require.IsType(t, &kvpb.ConditionFailedError{}, err)
-					return err
 				} else {
 					require.NoError(t, err)
 				}
@@ -2025,10 +2043,11 @@ func TestTxnBufferedWritesConditionalPuts(t *testing.T) {
 				}
 			})
 
-			if commit && !expErr {
+			if commit {
 				require.NoError(t, err)
 			} else {
 				require.Error(t, err)
+				testutils.IsError(err, "abort")
 			}
 
 			// Verify the values are visible only if the transaction commited