clear RPC error only if prewrite succeeds (#187)

Signed-off-by: Yilin Chen <sticnarf@gmail.com>

integration_tests/async_commit_fail_test.go

@@ -259,51 +259,3 @@ func (s *testAsyncCommitFailSuite) TestAsyncCommitContextCancelCausingUndetermin
 	s.NotNil(err)
 	s.NotNil(txn.GetCommitter().GetUndeterminedErr())
 }
-
-// TestAsyncCommitRPCErrorThenWriteConflict verifies that the determined failure error overwrites undetermined error.
-func (s *testAsyncCommitFailSuite) TestAsyncCommitRPCErrorThenWriteConflict() {
-	// This test doesn't support tikv mode because it needs setting failpoint in unistore.
-	if *mockstore.WithTiKV {
-		return
-	}
-
-	txn := s.beginAsyncCommit()
-	err := txn.Set([]byte("a"), []byte("va"))
-	s.Nil(err)
-
-	s.Nil(failpoint.Enable("tikvclient/rpcPrewriteResult", `1*return("timeout")->return("writeConflict")`))
-	defer func() {
-		s.Nil(failpoint.Disable("tikvclient/rpcPrewriteResult"))
-	}()
-
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
-	err = txn.Commit(ctx)
-	s.NotNil(err)
-	s.Nil(txn.GetCommitter().GetUndeterminedErr())
-}
-
-// TestAsyncCommitRPCErrorThenWriteConflictInChild verifies that the determined failure error in a child recursion
-// overwrites the undetermined error in the parent.
-func (s *testAsyncCommitFailSuite) TestAsyncCommitRPCErrorThenWriteConflictInChild() {
-	// This test doesn't support tikv mode because it needs setting failpoint in unistore.
-	if *mockstore.WithTiKV {
-		return
-	}
-
-	txn := s.beginAsyncCommit()
-	err := txn.Set([]byte("a"), []byte("va"))
-	s.Nil(err)
-
-	s.Nil(failpoint.Enable("tikvclient/rpcPrewriteResult", `1*return("timeout")->return("writeConflict")`))
-	s.Nil(failpoint.Enable("tikvclient/forceRecursion", `return`))
-
-	defer func() {
-		s.Nil(failpoint.Disable("tikvclient/rpcPrewriteResult"))
-		s.Nil(failpoint.Disable("tikvclient/forceRecursion"))
-	}()
-
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
-	err = txn.Commit(ctx)
-	s.NotNil(err)
-	s.Nil(txn.GetCommitter().GetUndeterminedErr())
-}

internal/locate/region_request.go

@@ -584,8 +584,6 @@ func (s *RegionRequestSender) SendReqCtx(
 				continue
 			}
 		} else {
-			// Clear the RPC Error since the request is evaluated successfully on a store.
-			s.rpcError = nil
 			if s.leaderReplicaSelector != nil {
 				s.leaderReplicaSelector.OnSendSuccess()
 			}

internal/locate/region_request_test.go

@@ -163,8 +163,6 @@ func (s *testRegionRequestToSingleStoreSuite) TestOnSendFailedWithStoreRestart()
 	resp, err = s.regionRequestSender.SendReq(s.bo, req, region.Region, time.Second)
 	s.Nil(err)
 	s.NotNil(resp.Resp)
-	// The RPC error should be nil since it's evaluated successfully.
-	s.Nil(s.regionRequestSender.rpcError)
 }
 
 func (s *testRegionRequestToSingleStoreSuite) TestOnSendFailedWithCloseKnownStoreThenUseNewOne() {

tikv/2pc.go

@@ -120,7 +120,6 @@ type twoPhaseCommitter struct {
 	maxCommitTS       uint64
 	prewriteStarted   bool
 	prewriteCancelled uint32
-	prewriteFailed    uint32
 	useOnePC          uint32
 	onePCCommitTS     uint64
 
@@ -1126,13 +1125,10 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) {
 	start := time.Now()
 	err = c.prewriteMutations(bo, c.mutations)
 
-	// Return an undetermined error only if we don't know the transaction fails.
-	// If it fails due to a write conflict or a already existed unique key, we
-	// needn't return an undetermined error even if such an error is set.
-	if atomic.LoadUint32(&c.prewriteFailed) == 1 {
-		c.setUndeterminedErr(nil)
-	}
 	if err != nil {
+		// TODO: Now we return an undetermined error as long as one of the prewrite
+		// RPCs fails. However, if there are multiple errors and some of the errors
+		// are not RPC failures, we can return the actual error instead of undetermined.
 		if undeterminedErr := c.getUndeterminedErr(); undeterminedErr != nil {
 			logutil.Logger(ctx).Error("2PC commit result undetermined",
 				zap.Error(err),

tikv/prewrite.go

@@ -223,9 +223,6 @@ func (action actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *Backoff
 			if err != nil {
 				return errors.Trace(err)
 			}
-			if _, err := util.EvalFailpoint("forceRecursion"); err == nil {
-				same = false
-			}
 			if same {
 				continue
 			}
@@ -239,6 +236,9 @@ func (action actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *Backoff
 		prewriteResp := resp.Resp.(*kvrpcpb.PrewriteResponse)
 		keyErrs := prewriteResp.GetErrors()
 		if len(keyErrs) == 0 {
+			// Clear the RPC Error since the request is evaluated successfully.
+			sender.SetRPCError(nil)
+
 			if batch.isPrimary {
 				// After writing the primary key, if the size of the transaction is larger than 32M,
 				// start the ttlManager. The ttlManager will be closed in tikvTxn.Commit().
@@ -299,17 +299,12 @@ func (action actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *Backoff
 			// Check already exists error
 			if alreadyExist := keyErr.GetAlreadyExist(); alreadyExist != nil {
 				e := &tikverr.ErrKeyExist{AlreadyExist: alreadyExist}
-				err = c.extractKeyExistsErr(e)
+				return c.extractKeyExistsErr(e)
-				if err != nil {
-					atomic.StoreUint32(&c.prewriteFailed, 1)
-				}
-				return err
 			}
 
 			// Extract lock from key error
 			lock, err1 := extractLockFromKeyErr(keyErr)
 			if err1 != nil {
-				atomic.StoreUint32(&c.prewriteFailed, 1)
 				return errors.Trace(err1)
 			}
 			logutil.BgLogger().Info("prewrite encounters lock",