store/tikv: fix lockTTL too large if local time is behind timestamp

store/tikv/2pc.go

@@ -708,6 +708,8 @@ func (tm *ttlManager) keepAlive(c *twoPhaseCommitter) {
 			}
 
 			newTTL := uptime + PessimisticLockTTL
+			logutil.Logger(context.Background()).Info("send TxnHeartBeat",
+				zap.Uint64("startTS", c.startTS), zap.Uint64("newTTL", newTTL))
 			startTime := time.Now()
 			_, err = sendTxnHeartBeat(bo, c.store, c.primary(), c.startTS, newTTL)
 			if err != nil {
@@ -736,8 +738,7 @@ func (action actionPessimisticLock) handleSingleBatch(c *twoPhaseCommitter, bo *
 		mutations[i] = mut
 	}
 
-	t0 := oracle.GetTimeFromTS(c.forUpdateTS)
-	elapsed := uint64(time.Since(t0) / time.Millisecond)
+	elapsed := uint64(time.Since(c.txn.startTime) / time.Millisecond)
 	req := &tikvrpc.Request{
 		Type: tikvrpc.CmdPessimisticLock,
 		PessimisticLock: &pb.PessimisticLockRequest{

store/tikv/2pc_test.go

@@ -570,6 +570,21 @@ func (s *testCommitterSuite) TestPessimisticTTL(c *C) {
 	c.Assert(false, IsTrue, Commentf("update pessimistic ttl fail"))
 }
 
+// TestElapsedTTL tests that elapsed time is correct even if ts physical time is greater than local time.
+func (s *testCommitterSuite) TestElapsedTTL(c *C) {
+	key := kv.Key("key")
+	txn := s.begin(c)
+	txn.startTS = oracle.ComposeTS(oracle.GetPhysical(time.Now().Add(time.Second*10)), 1)
+	txn.SetOption(kv.Pessimistic, true)
+	time.Sleep(time.Millisecond * 100)
+	forUpdateTS := oracle.ComposeTS(oracle.ExtractPhysical(txn.startTS)+100, 1)
+	err := txn.LockKeys(context.Background(), nil, forUpdateTS, kv.LockAlwaysWait, key)
+	c.Assert(err, IsNil)
+	lockInfo := s.getLockInfo(c, key)
+	c.Assert(lockInfo.LockTtl-PessimisticLockTTL, GreaterEqual, uint64(100))
+	c.Assert(lockInfo.LockTtl-PessimisticLockTTL, Less, uint64(150))
+}
+
 func (s *testCommitterSuite) getLockInfo(c *C, key []byte) *kvrpcpb.LockInfo {
 	txn := s.begin(c)
 	err := txn.Set(key, key)