func TestBadRequest(t *testing.T) {
	defer leaktest.AfterTest(t)
	s := server.StartTestServer(t)
	db := createTestClient(t, s.Stopper(), s.ServingAddr())
	defer s.Stop()

	// Write key "a".
	if pErr := db.Put("a", "value"); pErr != nil {
		t.Fatal(pErr)
	}

	if _, pErr := db.Scan("a", "a", 0); !testutils.IsPError(pErr, "truncation resulted in empty batch") {
		t.Fatalf("unexpected error on scan with startkey == endkey: %v", pErr)
	}

	if _, pErr := db.ReverseScan("a", "a", 0); !testutils.IsPError(pErr, "truncation resulted in empty batch") {
		t.Fatalf("unexpected pError on reverse scan with startkey == endkey: %v", pErr)
	}

	if pErr := db.DelRange("x", "a"); !testutils.IsPError(pErr, "truncation resulted in empty batch") {
		t.Fatalf("unexpected error on deletion on [x, a): %v", pErr)
	}

	if pErr := db.DelRange("", "z"); !testutils.IsPError(pErr, "must be greater than LocalMax") {
		t.Fatalf("unexpected error on deletion on [KeyMin, z): %v", pErr)
	}
}

// TestAuthentication tests authentication for the KV endpoint.
func TestAuthentication(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s := server.StartTestServer(t)
	defer s.Stop()

	var b1 client.Batch
	b1.Put("a", "b")

	// Create a node user client and call Run() on it which lets us build our own
	// request, specifying the user.
	db1 := createTestClientForUser(t, s.Stopper(), s.ServingAddr(), security.NodeUser)
	if pErr := db1.Run(&b1); pErr != nil {
		t.Fatal(pErr)
	}

	var b2 client.Batch
	b2.Put("c", "d")

	// Try again, but this time with certs for a non-node user (even the root
	// user has no KV permissions).
	db2 := createTestClientForUser(t, s.Stopper(), s.ServingAddr(), security.RootUser)
	if pErr := db2.Run(&b2); !testutils.IsPError(pErr, "is not allowed") {
		t.Fatal(pErr)
	}
}

func TestTxnAbortCount(t *testing.T) {
	defer leaktest.AfterTest(t)()
	_, sender, cleanupFn := setupMetricsTest(t)
	defer cleanupFn()

	value := []byte("value")
	db := client.NewDB(sender)

	intentionalErrText := "intentional error to cause abort"
	// Test aborted transaction.
	if pErr := db.Txn(func(txn *client.Txn) *roachpb.Error {
		key := []byte("key-abort")

		if err := txn.SetIsolation(roachpb.SNAPSHOT); err != nil {
			return roachpb.NewError(err)
		}

		if pErr := txn.Put(key, value); pErr != nil {
			t.Fatal(pErr)
		}

		return roachpb.NewErrorf(intentionalErrText)
	}); !testutils.IsPError(pErr, intentionalErrText) {
		t.Fatalf("unexpected error: %s", pErr)
	}
	teardownHeartbeats(sender)
	checkTxnMetrics(t, sender, "abort txn", 0, 0, 1, 0)
}

// TestTxnCoordSenderErrorWithIntent validates that if a transactional request
// returns an error but also indicates a Writing transaction, the coordinator
// tracks it just like a successful request.
func TestTxnCoordSenderErrorWithIntent(t *testing.T) {
	defer leaktest.AfterTest(t)
	stopper := stop.NewStopper()
	manual := hlc.NewManualClock(0)
	clock := hlc.NewClock(manual.UnixNano)
	clock.SetMaxOffset(20)

	ts := NewTxnCoordSender(senderFn(func(_ context.Context, ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
		txn := ba.Txn.Clone()
		txn.Writing = true
		pErr := roachpb.NewError(roachpb.NewTransactionRetryError())
		pErr.SetTxn(txn)
		return nil, pErr
	}), clock, false, nil, stopper)
	defer stopper.Stop()

	var ba roachpb.BatchRequest
	key := roachpb.Key("test")
	ba.Add(&roachpb.BeginTransactionRequest{Span: roachpb.Span{Key: key}})
	ba.Add(&roachpb.PutRequest{Span: roachpb.Span{Key: key}})
	ba.Add(&roachpb.EndTransactionRequest{})
	ba.Txn = &roachpb.Transaction{Name: "test"}
	if _, pErr := ts.Send(context.Background(), ba); !testutils.IsPError(pErr, "retry txn") {
		t.Fatalf("unexpected error: %v", pErr)
	}

	defer teardownHeartbeats(ts)
	ts.Lock()
	defer ts.Unlock()
	if len(ts.txns) != 1 {
		t.Fatalf("expected transaction to be tracked")
	}
}

// Verifies that an inner transaction in a nested transaction strips the transaction
// information in its error when propagating it to an other transaction.
func TestNestedTransaction(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s, db := setup()
	defer s.Stop()

	pErr := db.Txn(func(txn1 *client.Txn) *roachpb.Error {
		if pErr := txn1.Put("a", "1"); pErr != nil {
			t.Fatalf("unexpected put error: %s", pErr)
		}
		iPErr := db.Txn(func(txn2 *client.Txn) *roachpb.Error {
			txnProto := roachpb.NewTransaction("test", roachpb.Key("a"), 1, roachpb.SERIALIZABLE, roachpb.Timestamp{}, 0)
			return roachpb.NewErrorWithTxn(util.Errorf("inner txn error"), txnProto)
		})

		if iPErr.GetTxn() != nil {
			t.Errorf("error txn must be stripped: %s", iPErr)
		}
		return iPErr

	})
	if pErr == nil {
		t.Fatal("unexpected success of txn")
	}
	if !testutils.IsPError(pErr, "inner txn error") {
		t.Errorf("unexpected failure: %s", pErr)
	}
}

func TestTxnAbandonCount(t *testing.T) {
	defer leaktest.AfterTest(t)()
	manual, sender, cleanupFn := setupMetricsTest(t)
	defer cleanupFn()
	value := []byte("value")
	db := client.NewDB(sender)

	// Test abandoned transaction by making the client timeout ridiculously short. We also set
	// the sender to heartbeat very frequently, because the heartbeat detects and tears down
	// abandoned transactions.
	sender.heartbeatInterval = 2 * time.Millisecond
	sender.clientTimeout = 1 * time.Millisecond
	if pErr := db.Txn(func(txn *client.Txn) *roachpb.Error {
		key := []byte("key-abandon")

		if err := txn.SetIsolation(roachpb.SNAPSHOT); err != nil {
			return roachpb.NewError(err)
		}

		if pErr := txn.Put(key, value); pErr != nil {
			return pErr
		}

		manual.Increment(int64(sender.clientTimeout + sender.heartbeatInterval*2))

		checkTxnMetrics(t, sender, "abandon txn", 0, 1, 0, 0)

		return nil
	}); !testutils.IsPError(pErr, "already committed or aborted") {
		t.Fatalf("unexpected error: %s", pErr)
	}
}

// TestTxnDBBasics verifies that a simple transaction can be run and
// either committed or aborted. On commit, mutations are visible; on
// abort, mutations are never visible. During the txn, verify that
// uncommitted writes cannot be read outside of the txn but can be
// read from inside the txn.
func TestTxnDBBasics(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s := createTestDB(t)
	defer s.Stop()
	value := []byte("value")

	for _, commit := range []bool{true, false} {
		key := []byte(fmt.Sprintf("key-%t", commit))

		pErr := s.DB.Txn(func(txn *client.Txn) *roachpb.Error {
			// Use snapshot isolation so non-transactional read can always push.
			if err := txn.SetIsolation(roachpb.SNAPSHOT); err != nil {
				return roachpb.NewError(err)
			}

			// Put transactional value.
			if pErr := txn.Put(key, value); pErr != nil {
				return pErr
			}

			// Attempt to read outside of txn.
			if gr, pErr := s.DB.Get(key); pErr != nil {
				return pErr
			} else if gr.Exists() {
				return roachpb.NewErrorf("expected nil value; got %v", gr.Value)
			}

			// Read within the transaction.
			if gr, pErr := txn.Get(key); pErr != nil {
				return pErr
			} else if !gr.Exists() || !bytes.Equal(gr.ValueBytes(), value) {
				return roachpb.NewErrorf("expected value %q; got %q", value, gr.Value)
			}

			if !commit {
				return roachpb.NewErrorf("purposefully failing transaction")
			}
			return nil
		})

		if commit != (pErr == nil) {
			t.Errorf("expected success? %t; got %s", commit, pErr)
		} else if !commit && !testutils.IsPError(pErr, "purposefully failing transaction") {
			t.Errorf("unexpected failure with !commit: %s", pErr)
		}

		// Verify the value is now visible on commit == true, and not visible otherwise.
		gr, pErr := s.DB.Get(key)
		if commit {
			if pErr != nil || !gr.Exists() || !bytes.Equal(gr.ValueBytes(), value) {
				t.Errorf("expected success reading value: %+v, %s", gr.ValueBytes(), pErr)
			}
		} else {
			if pErr != nil || gr.Exists() {
				t.Errorf("expected success and nil value: %s, %s", gr, pErr)
			}
		}
	}
}

// TestClientRunTransaction verifies some simple transaction isolation
// semantics.
func TestClientRunTransaction(t *testing.T) {
	defer leaktest.AfterTest(t)
	s := server.StartTestServer(t)
	defer s.Stop()
	defer setTxnRetryBackoff(1 * time.Millisecond)()
	db := createTestClient(t, s.Stopper(), s.ServingAddr())

	for _, commit := range []bool{true, false} {
		value := []byte("value")
		key := []byte(fmt.Sprintf("%s/key-%t", testUser, commit))

		// Use snapshot isolation so non-transactional read can always push.
		pErr := db.Txn(func(txn *client.Txn) *roachpb.Error {
			if pErr := txn.SetIsolation(roachpb.SNAPSHOT); pErr != nil {
				return pErr
			}

			// Put transactional value.
			if pErr := txn.Put(key, value); pErr != nil {
				return pErr
			}
			// Attempt to read outside of txn.
			if gr, pErr := db.Get(key); pErr != nil {
				return pErr
			} else if gr.Value != nil {
				return roachpb.NewErrorf("expected nil value; got %+v", gr.Value)
			}
			// Read within the transaction.
			if gr, pErr := txn.Get(key); pErr != nil {
				return pErr
			} else if gr.Value == nil || !bytes.Equal(gr.ValueBytes(), value) {
				return roachpb.NewErrorf("expected value %q; got %q", value, gr.ValueBytes())
			}
			if !commit {
				return roachpb.NewErrorf("purposefully failing transaction")
			}
			return nil
		})

		if commit != (pErr == nil) {
			t.Errorf("expected success? %t; got %s", commit, pErr)
		} else if !commit && !testutils.IsPError(pErr, "purposefully failing transaction") {
			t.Errorf("unexpected failure with !commit: %s", pErr)
		}

		// Verify the value is now visible on commit == true, and not visible otherwise.
		gr, pErr := db.Get(key)
		if commit {
			if pErr != nil || gr.Value == nil || !bytes.Equal(gr.ValueBytes(), value) {
				t.Errorf("expected success reading value: %+v, %s", gr.Value, pErr)
			}
		} else {
			if pErr != nil || gr.Value != nil {
				t.Errorf("expected success and nil value: %+v, %s", gr.Value, pErr)
			}
		}
	}
}

// TestClientPermissions verifies permission enforcement.
func TestClientPermissions(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s := server.StartTestServer(t)
	defer s.Stop()

	// NodeUser certs are required for all KV operations.
	// RootUser has no KV privileges whatsoever.
	nodeClient := createTestClientForUser(t, s.Stopper(), s.ServingAddr(), security.NodeUser)
	rootClient := createTestClientForUser(t, s.Stopper(), s.ServingAddr(), security.RootUser)

	testCases := []struct {
		path    string
		client  *client.DB
		allowed bool
	}{
		{"foo", rootClient, false},
		{"foo", nodeClient, true},

		{testUser + "/foo", rootClient, false},
		{testUser + "/foo", nodeClient, true},

		{testUser + "foo", rootClient, false},
		{testUser + "foo", nodeClient, true},

		{testUser, rootClient, false},
		{testUser, nodeClient, true},

		{"unknown/foo", rootClient, false},
		{"unknown/foo", nodeClient, true},
	}

	value := []byte("value")
	const matchErr = "is not allowed"
	for tcNum, tc := range testCases {
		pErr := tc.client.Put(tc.path, value)
		if (pErr == nil) != tc.allowed || (!tc.allowed && !testutils.IsPError(pErr, matchErr)) {
			t.Errorf("#%d: expected allowed=%t, got err=%s", tcNum, tc.allowed, pErr)
		}
		_, pErr = tc.client.Get(tc.path)
		if (pErr == nil) != tc.allowed || (!tc.allowed && !testutils.IsPError(pErr, matchErr)) {
			t.Errorf("#%d: expected allowed=%t, got err=%s", tcNum, tc.allowed, pErr)
		}
	}
}

// TestStoreRangeMergeLastRange verifies that merging the last range
// fails.
func TestStoreRangeMergeLastRange(t *testing.T) {
	defer leaktest.AfterTest(t)
	store, stopper := createTestStore(t)
	defer stopper.Stop()

	// Merge last range.
	args := adminMergeArgs(roachpb.KeyMin)
	if _, pErr := client.SendWrapped(rg1(store), nil, &args); !testutils.IsPError(pErr, "cannot merge final range") {
		t.Fatalf("expected 'cannot merge final range' error; got %s", pErr)
	}
}

// TestTxnCoordSenderErrorWithIntent validates that if a transactional request
// returns an error but also indicates a Writing transaction, the coordinator
// tracks it just like a successful request.
func TestTxnCoordSenderErrorWithIntent(t *testing.T) {
	defer leaktest.AfterTest(t)()
	stopper := stop.NewStopper()
	defer stopper.Stop()
	manual := hlc.NewManualClock(0)
	clock := hlc.NewClock(manual.UnixNano)
	clock.SetMaxOffset(20)

	testCases := []struct {
		roachpb.Error
		errMsg string
	}{
		{*roachpb.NewError(roachpb.NewTransactionRetryError()), "retry txn"},
		{*roachpb.NewError(roachpb.NewTransactionPushError(roachpb.Transaction{
			TxnMeta: enginepb.TxnMeta{
				ID: uuid.NewV4(),
			}})), "failed to push"},
		{*roachpb.NewErrorf("testError"), "testError"},
	}
	for i, test := range testCases {
		func() {
			senderFunc := func(_ context.Context, ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
				txn := ba.Txn.Clone()
				txn.Writing = true
				pErr := &roachpb.Error{}
				*pErr = test.Error
				pErr.SetTxn(&txn)
				return nil, pErr
			}
			ctx := tracing.WithTracer(context.Background(), tracing.NewTracer())
			ts := NewTxnCoordSender(ctx, senderFn(senderFunc), clock, false, stopper, MakeTxnMetrics())

			var ba roachpb.BatchRequest
			key := roachpb.Key("test")
			ba.Add(&roachpb.BeginTransactionRequest{Span: roachpb.Span{Key: key}})
			ba.Add(&roachpb.PutRequest{Span: roachpb.Span{Key: key}})
			ba.Add(&roachpb.EndTransactionRequest{})
			ba.Txn = &roachpb.Transaction{Name: "test"}
			_, pErr := ts.Send(context.Background(), ba)
			if !testutils.IsPError(pErr, test.errMsg) {
				t.Errorf("%d: error did not match %s: %v", i, test.errMsg, pErr)
			}

			defer teardownHeartbeats(ts)
			ts.Lock()
			defer ts.Unlock()
			if len(ts.txns) != 1 {
				t.Errorf("%d: expected transaction to be tracked", i)
			}
		}()
	}
}

// TestStoreRangeMergeLastRange verifies that merging the last range
// fails.
func TestStoreRangeMergeLastRange(t *testing.T) {
	defer leaktest.AfterTest(t)()
	sCtx := storage.TestStoreContext()
	sCtx.TestingKnobs.DisableSplitQueue = true
	store, stopper, _ := createTestStoreWithContext(t, sCtx)
	defer stopper.Stop()

	// Merge last range.
	args := adminMergeArgs(roachpb.KeyMin)
	if _, pErr := client.SendWrapped(rg1(store), nil, &args); !testutils.IsPError(pErr, "cannot merge final range") {
		t.Fatalf("expected 'cannot merge final range' error; got %s", pErr)
	}
}

// TestPushTransactionsWithNonPendingIntent verifies that maybePushTransactions
// returns an error when a non-pending intent is passed.
func TestPushTransactionsWithNonPendingIntent(t *testing.T) {
	defer leaktest.AfterTest(t)()

	tc := testContext{}
	tc.Start(t)
	defer tc.Stop()

	intents := []roachpb.Intent{{Span: roachpb.Span{Key: roachpb.Key("a")}, Status: roachpb.ABORTED}}
	if _, pErr := tc.store.intentResolver.maybePushTransactions(
		context.Background(), intents, roachpb.Header{}, roachpb.PUSH_TOUCH, true); !testutils.IsPError(pErr, "unexpected aborted/resolved intent") {
		t.Errorf("expected error on aborted/resolved intent, but got %s", pErr)
	}
}

// TestStoreRangeMergeNonCollocated attempts to merge two ranges
// that are not on the same stores.
func TestStoreRangeMergeNonCollocated(t *testing.T) {
	defer leaktest.AfterTest(t)()
	mtc := startMultiTestContext(t, 4)
	defer mtc.Stop()

	store := mtc.stores[0]

	// Split into 3 ranges
	argsSplit := adminSplitArgs(roachpb.KeyMin, []byte("d"))
	if _, pErr := client.SendWrapped(rg1(store), nil, &argsSplit); pErr != nil {
		t.Fatalf("Can't split range %s", pErr)
	}
	argsSplit = adminSplitArgs(roachpb.KeyMin, []byte("b"))
	if _, pErr := client.SendWrapped(rg1(store), nil, &argsSplit); pErr != nil {
		t.Fatalf("Can't split range %s", pErr)
	}

	rangeA := store.LookupReplica([]byte("a"), nil)
	rangeADesc := rangeA.Desc()
	rangeB := store.LookupReplica([]byte("c"), nil)
	rangeBDesc := rangeB.Desc()
	rangeC := store.LookupReplica([]byte("e"), nil)
	rangeCDesc := rangeC.Desc()

	if bytes.Equal(rangeADesc.StartKey, rangeBDesc.StartKey) {
		log.Errorf(context.TODO(), "split ranges keys are equal %q!=%q", rangeADesc.StartKey, rangeBDesc.StartKey)
	}
	if bytes.Equal(rangeBDesc.StartKey, rangeCDesc.StartKey) {
		log.Errorf(context.TODO(), "split ranges keys are equal %q!=%q", rangeBDesc.StartKey, rangeCDesc.StartKey)
	}
	if bytes.Equal(rangeADesc.StartKey, rangeCDesc.StartKey) {
		log.Errorf(context.TODO(), "split ranges keys are equal %q!=%q", rangeADesc.StartKey, rangeCDesc.StartKey)
	}

	// Replicate the ranges to different sets of stores. Ranges A and C
	// are collocated, but B is different.
	mtc.replicateRange(rangeA.RangeID, 1, 2)
	mtc.replicateRange(rangeB.RangeID, 1, 3)
	mtc.replicateRange(rangeC.RangeID, 1, 2)

	// Attempt to merge.
	rangeADesc = rangeA.Desc()
	argsMerge := adminMergeArgs(roachpb.Key(rangeADesc.StartKey))
	if _, pErr := rangeA.AdminMerge(context.Background(), argsMerge, rangeADesc); !testutils.IsPError(pErr, "ranges not collocated") {
		t.Fatalf("did not got expected error; got %s", pErr)
	}
}

// TestStoreRangeSplitAtIllegalKeys verifies a range cannot be split
// at illegal keys.
func TestStoreRangeSplitAtIllegalKeys(t *testing.T) {
	defer leaktest.AfterTest(t)()
	store, stopper, _ := createTestStore(t)
	defer stopper.Stop()

	for _, key := range []roachpb.Key{
		keys.Meta1Prefix,
		testutils.MakeKey(keys.Meta1Prefix, []byte("a")),
		testutils.MakeKey(keys.Meta1Prefix, roachpb.RKeyMax),
		keys.Meta2KeyMax,
		keys.MakeTablePrefix(10 /* system descriptor ID */),
	} {
		args := adminSplitArgs(roachpb.KeyMin, key)
		_, pErr := client.SendWrapped(rg1(store), nil, &args)
		if !testutils.IsPError(pErr, "cannot split") {
			t.Errorf("%q: unexpected split error %s", key, pErr)
		}
	}
}

func TestKVDBInternalMethods(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s := server.StartTestServer(t)
	defer s.Stop()

	testCases := []roachpb.Request{
		&roachpb.HeartbeatTxnRequest{},
		&roachpb.GCRequest{},
		&roachpb.PushTxnRequest{},
		&roachpb.RangeLookupRequest{},
		&roachpb.ResolveIntentRequest{},
		&roachpb.ResolveIntentRangeRequest{},
		&roachpb.MergeRequest{},
		&roachpb.TruncateLogRequest{},
		&roachpb.LeaderLeaseRequest{},

		&roachpb.EndTransactionRequest{
			InternalCommitTrigger: &roachpb.InternalCommitTrigger{},
		},
	}
	// Verify internal methods experience bad request errors.
	db := createTestClient(t, s.Stopper(), s.ServingAddr())
	for i, args := range testCases {
		{
			header := args.Header()
			header.Key = roachpb.Key("a")
			args.SetHeader(header)
		}
		if roachpb.IsRange(args) {
			header := args.Header()
			header.EndKey = args.Header().Key.Next()
			args.SetHeader(header)
		}
		b := &client.Batch{}
		b.InternalAddRequest(args)
		pErr := db.Run(b)
		if pErr == nil {
			t.Errorf("%d: unexpected success calling %s", i, args.Method())
		} else if !testutils.IsPError(pErr, "contains an internal request|contains commit trigger") {
			t.Errorf("%d: unexpected error for %s: %s", i, args.Method(), pErr)
		}
	}
}

func TestPrimaryKeyUnspecified(t *testing.T) {
	defer leaktest.AfterTest(t)
	stmt, err := parser.ParseOneTraditional(
		"CREATE TABLE foo.test (a INT, b INT, CONSTRAINT c UNIQUE (b))")
	if err != nil {
		t.Fatal(err)
	}
	create := stmt.(*parser.CreateTable)
	if err := create.Table.NormalizeTableName(""); err != nil {
		t.Fatal(err)
	}
	desc, pErr := makeTableDesc(create, 1)
	if pErr != nil {
		t.Fatal(pErr)
	}
	pErr = desc.AllocateIDs()
	if !testutils.IsPError(pErr, errMissingPrimaryKey.Error()) {
		t.Fatalf("unexpected error: %s", pErr)
	}
}

// TestTxnReadAfterAbandon checks the fix for the condition in issue #4787:
// after a transaction is abandoned we do a read as part of that transaction
// which should fail.
func TestTxnReadAfterAbandon(t *testing.T) {
	defer leaktest.AfterTest(t)()
	manual, sender, cleanupFn := setupMetricsTest(t)
	defer cleanupFn()
	value := []byte("value")
	db := client.NewDB(sender)

	// Test abandoned transaction by making the client timeout ridiculously short. We also set
	// the sender to heartbeat very frequently, because the heartbeat detects and tears down
	// abandoned transactions.
	sender.heartbeatInterval = 2 * time.Millisecond
	sender.clientTimeout = 1 * time.Millisecond

	pErr := db.Txn(func(txn *client.Txn) *roachpb.Error {
		key := []byte("key-abandon")

		if err := txn.SetIsolation(roachpb.SNAPSHOT); err != nil {
			t.Fatal(err)
		}

		if pErr := txn.Put(key, value); pErr != nil {
			t.Fatal(pErr)
		}

		manual.Increment(int64(sender.clientTimeout + sender.heartbeatInterval*2))

		checkTxnMetrics(t, sender, "abandon txn", 0, 1, 0, 0)

		_, pErr := txn.Get(key)
		if pErr == nil {
			t.Fatalf("Get succeeded on abandoned txn")
		} else if !testutils.IsPError(pErr, "writing transaction timed out") {
			t.Fatalf("unexpected error from Get on abandoned txn: %s", pErr)
		}
		return pErr
	})

	if pErr == nil {
		t.Fatalf("abandoned txn didn't fail")
	}
}

// Verifies that a nested transaction returns an error if an inner txn
// propagates an error to an outer txn.
func TestNestedTransaction(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s, db := setup()
	defer s.Stop()

	txnProto := roachpb.NewTransaction("test", roachpb.Key("a"), 1, roachpb.SERIALIZABLE, roachpb.Timestamp{}, 0)
	pErr := db.Txn(func(txn1 *client.Txn) *roachpb.Error {
		if pErr := txn1.Put("a", "1"); pErr != nil {
			t.Fatalf("unexpected put error: %s", pErr)
		}
		return db.Txn(func(txn2 *client.Txn) *roachpb.Error {
			return roachpb.NewErrorWithTxn(util.Errorf("err"), txnProto)
		})
	})
	if pErr == nil {
		t.Fatal("unexpected success of txn")
	}
	if !testutils.IsPError(pErr, "mismatching transaction record in the error") {
		t.Errorf("unexpected failure: %s", pErr)
	}
}

// TestTxnCoordIdempotentCleanup verifies that cleanupTxn is idempotent.
func TestTxnCoordIdempotentCleanup(t *testing.T) {
	defer leaktest.AfterTest(t)()
	s := createTestDB(t)
	defer s.Stop()
	defer teardownHeartbeats(s.Sender)

	txn := client.NewTxn(*s.DB)
	ba := txn.NewBatch()
	ba.Put(roachpb.Key("a"), []byte("value"))
	if pErr := txn.Run(ba); pErr != nil {
		t.Fatal(pErr)
	}

	s.Sender.cleanupTxn(context.Background(), txn.Proto)

	ba = txn.NewBatch()
	ba.InternalAddRequest(&roachpb.EndTransactionRequest{})
	pErr := txn.Run(ba)
	if pErr != nil && !testutils.IsPError(pErr, "aborted") {
		t.Fatal(pErr)
	}
}