Update TTL in heartbeat request (#280)

* Update TTL in heartbeat request

Signed-off-by: Zijie Lu <[email protected]>

* Fix TTL calculation

Signed-off-by: Andy Lok <[email protected]>

* Fix clippy.

Signed-off-by: Zijie Lu <[email protected]>

* Update transaction.rs

Signed-off-by: Andy Lok <[email protected]>

* Update transaction.rs

Signed-off-by: Andy Lok <[email protected]>

* style: fix clippy

Signed-off-by: ekexium <[email protected]>

Co-authored-by: Andy Lok <[email protected]>
Co-authored-by: Ziqian Qin <[email protected]>

Author: 3 people

src/kv/kvpair.rs

@@ -133,9 +133,9 @@ impl AsRef<Value> for KvPair {
 impl fmt::Debug for KvPair {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         let KvPair(key, value) = self;
-        match str::from_utf8(&value) {
+        match str::from_utf8(value) {
             Ok(s) => write!(f, "KvPair({}, {:?})", HexRepr(&key.0), s),
-            Err(_) => write!(f, "KvPair({}, {})", HexRepr(&key.0), HexRepr(&value)),
+            Err(_) => write!(f, "KvPair({}, {})", HexRepr(&key.0), HexRepr(value)),
         }
     }
 }

src/transaction/buffer.rs

@@ -237,7 +237,7 @@ impl Buffer {
 
     fn get_from_mutations(&self, key: &Key) -> MutationValue {
         self.entry_map
-            .get(&key)
+            .get(key)
             .map(BufferEntry::get_value)
             .unwrap_or(MutationValue::Undetermined)
     }

src/transaction/transaction.rs

@@ -646,7 +646,11 @@ impl<PdC: PdClient> Transaction<PdC> {
             Some(k) => k,
             None => return Err(Error::NoPrimaryKey),
         };
-        let request = new_heart_beat_request(self.timestamp.clone(), primary_key, DEFAULT_LOCK_TTL);
+        let request = new_heart_beat_request(
+            self.timestamp.clone(),
+            primary_key,
+            self.start_instant.elapsed().as_millis() as u64 + DEFAULT_LOCK_TTL,
+        );
         let plan = PlanBuilder::new(self.rpc.clone(), request)
             .single_region()
             .await?
@@ -787,6 +791,7 @@ impl<PdC: PdClient> Transaction<PdC> {
             HeartbeatOption::NoHeartbeat => DEFAULT_HEARTBEAT_INTERVAL,
             HeartbeatOption::FixedTime(heartbeat_interval) => heartbeat_interval,
         };
+        let start_instant = self.start_instant;
 
         let heartbeat_task = async move {
             loop {
@@ -802,11 +807,10 @@ impl<PdC: PdClient> Transaction<PdC> {
                         break;
                     }
                 }
-                let current_ts = rpc.clone().get_timestamp().await?;
                 let request = new_heart_beat_request(
                     start_ts.clone(),
                     primary_key.clone(),
-                    (current_ts.physical - start_ts.physical) as u64 + DEFAULT_LOCK_TTL,
+                    start_instant.elapsed().as_millis() as u64 + DEFAULT_LOCK_TTL,
                 );
                 let plan = PlanBuilder::new(rpc.clone(), request)
                     .single_region()