Merge pull request #248 from ekexium/reduce-panic

Reduce some panics

Author: nrc

src/region.rs

@@ -28,11 +28,6 @@ pub struct Region {
 }
 
 impl Region {
-    #[allow(dead_code)]
-    pub fn switch_peer(&mut self, _to: StoreId) -> Result<()> {
-        unimplemented!()
-    }
-
     pub fn contains(&self, key: &Key) -> bool {
         let key: &[u8] = key.into();
         let start_key = self.region.get_start_key();

src/request/plan.rs

@@ -40,11 +40,18 @@ impl<Req: KvRequest> Plan for Dispatch<Req> {
         let result = self
             .kv_client
             .as_ref()
-            .expect("Unreachable: kv_client has not been initialised in Dispatch")
+            .ok_or_else(|| {
+                Error::StringError(
+                    "Unreachable: kv_client has not been initialised in Dispatch".to_owned(),
+                )
+            })?
             .dispatch(&self.request)
             .await;
         let result = stats.done(result);
-        result.map(|r| *r.downcast().expect("Downcast failed"))
+        result.map(|r| {
+            *r.downcast()
+                .expect("Downcast failed: request and response type mismatch")
+        })
     }
 }
 

src/transaction/transaction.rs

@@ -628,7 +628,7 @@ impl Transaction {
         let first_key = keys[0].clone();
         let primary_lock = self.buffer.get_primary_key_or(&first_key).await;
         let lock_ttl = DEFAULT_LOCK_TTL;
-        let for_update_ts = self.rpc.clone().get_timestamp().await.unwrap();
+        let for_update_ts = self.rpc.clone().get_timestamp().await?;
         self.options.push_for_update_ts(for_update_ts.clone());
         let request = new_pessimistic_lock_request(
             keys.clone().into_iter(),
@@ -843,6 +843,7 @@ impl Committer {
         }
 
         let commit_ts = if self.options.async_commit {
+            // FIXME: min_commit_ts == 0 => fallback to normal 2PC
             min_commit_ts.unwrap()
         } else {
             match self.commit_primary().await {

tikv-client-pd/src/cluster.rs

@@ -290,7 +290,7 @@ impl PdMessage for pdpb::GetRegionRequest {
     type Response = pdpb::GetRegionResponse;
 
     async fn rpc(&self, client: &pdpb::PdClient, opt: CallOption) -> GrpcResult<Self::Response> {
-        client.get_region_async_opt(self, opt).unwrap().await
+        client.get_region_async_opt(self, opt)?.await
     }
 }
 
@@ -299,7 +299,7 @@ impl PdMessage for pdpb::GetRegionByIdRequest {
     type Response = pdpb::GetRegionResponse;
 
     async fn rpc(&self, client: &pdpb::PdClient, opt: CallOption) -> GrpcResult<Self::Response> {
-        client.get_region_by_id_async_opt(self, opt).unwrap().await
+        client.get_region_by_id_async_opt(self, opt)?.await
     }
 }
 
@@ -308,7 +308,7 @@ impl PdMessage for pdpb::GetStoreRequest {
     type Response = pdpb::GetStoreResponse;
 
     async fn rpc(&self, client: &pdpb::PdClient, opt: CallOption) -> GrpcResult<Self::Response> {
-        client.get_store_async_opt(self, opt).unwrap().await
+        client.get_store_async_opt(self, opt)?.await
     }
 }
 
@@ -317,7 +317,7 @@ impl PdMessage for pdpb::GetAllStoresRequest {
     type Response = pdpb::GetAllStoresResponse;
 
     async fn rpc(&self, client: &pdpb::PdClient, opt: CallOption) -> GrpcResult<Self::Response> {
-        client.get_all_stores_async_opt(self, opt).unwrap().await
+        client.get_all_stores_async_opt(self, opt)?.await
     }
 }
 
@@ -326,10 +326,7 @@ impl PdMessage for pdpb::UpdateGcSafePointRequest {
     type Response = pdpb::UpdateGcSafePointResponse;
 
     async fn rpc(&self, client: &pdpb::PdClient, opt: CallOption) -> GrpcResult<Self::Response> {
-        client
-            .update_gc_safe_point_async_opt(self, opt)
-            .unwrap()
-            .await
+        client.update_gc_safe_point_async_opt(self, opt)?.await
     }
 }