Problem (Fix #1881): wallet sync logic adapt to new tendermint-rs lig…

…ht client design

Solution:
- running light client superivsor, and sync wallet to recent
  trusted block

Cargo.toml

@@ -53,3 +53,16 @@ default-members = [
 
 [patch.crates-io]
 ring = { git = "https://github.com/crypto-com/ring.git", rev = "7d08bdbbc68fe5e867445c576c004336699650c1" }
+
+[patch."https://github.com/informalsystems/tendermint-rs.git".tendermint]
+git = "https://github.com/crypto-com/tendermint-rs.git"
+branch = "fix-serialization"
+default-features = false
+
+[patch."https://github.com/informalsystems/tendermint-rs.git".tendermint-rpc]
+git = "https://github.com/crypto-com/tendermint-rs.git"
+branch = "fix-serialization"
+
+[patch."https://github.com/informalsystems/tendermint-rs.git".tendermint-light-client]
+git = "https://github.com/crypto-com/tendermint-rs.git"
+branch = "fix-serialization"

chain-tx-enclave-next/enclave-ra/ra-client/src/verifier.rs

@@ -102,10 +102,10 @@ impl EnclaveCertVerifier {
         } = certificate.tbs_certificate.validity;
         let now_sec = now.timestamp();
 
-        if now_sec < not_before.to_timespec().sec {
+        if now_sec < not_before.timestamp() {
             return Err(EnclaveCertVerifierError::CertificateNotBegin);
         }
-        if now_sec >= not_after.to_timespec().sec {
+        if now_sec >= not_after.timestamp() {
             return Err(EnclaveCertVerifierError::CertificateExpired);
         }
 

client-cli/src/command.rs

@@ -28,7 +28,8 @@ use client_core::signer::WalletSignerManager;
 use client_core::transaction_builder::DefaultWalletTransactionBuilder;
 use client_core::types::BalanceChange;
 use client_core::wallet::syncer::{
-    ObfuscationSyncerConfig, ProgressReport, SyncerOptions, WalletSyncer,
+    spawn_light_client_supervisor, Handle, ObfuscationSyncerConfig, ProgressReport, SyncerOptions,
+    WalletSyncer,
 };
 use client_core::wallet::{DefaultWalletClient, WalletClient};
 use client_network::network_ops::{DefaultNetworkOpsClient, NetworkOpsClient};
@@ -385,10 +386,13 @@ impl Command {
                 block_height_ensure,
             } => {
                 let enckey = ask_seckey(None)?;
-                let tendermint_client = WebsocketRpcClient::new(&tendermint_url())?;
+                let rpc_url = tendermint_url();
+                let tendermint_client = WebsocketRpcClient::new(&rpc_url)?;
                 let tx_obfuscation = get_tx_query(tendermint_client.clone())?;
 
-                let storage = SledStorage::new(storage_path())?;
+                let db_path = storage_path();
+                let storage = SledStorage::new(&db_path)?;
+                let handle = spawn_light_client_supervisor(db_path.as_ref(), &rpc_url);
                 let config = ObfuscationSyncerConfig::new(
                     storage.clone(),
                     tendermint_client,
@@ -399,8 +403,12 @@ impl Command {
                         batch_size: *batch_size,
                         block_height_ensure: *block_height_ensure,
                     },
+                    handle.clone(),
                 );
                 Self::resync(config, name.clone(), enckey, *force, storage)?;
+                handle
+                    .terminate()
+                    .expect("terminate light client supervisor in client-cli");
                 Ok(())
             }
             Command::MultiSig { multisig_command } => {
@@ -637,8 +645,8 @@ impl Command {
         Ok(())
     }
 
-    fn resync<S: Storage, C: Client, O: TransactionObfuscation>(
-        config: ObfuscationSyncerConfig<S, C, O>,
+    fn resync<S: Storage, C: Client, O: TransactionObfuscation, L: Handle + Send + Sync + Clone>(
+        config: ObfuscationSyncerConfig<S, C, O, L>,
         name: String,
         enckey: SecKey,
         force: bool,

client-common/Cargo.toml

@@ -35,7 +35,8 @@ secstr = { version = "0.4.0", features = ["serde"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sled = { version = "0.31.0", optional = true }
-tendermint = { git = "https://github.com/crypto-com/tendermint-rs.git", default-features = false, rev = "e8d350960726b242fdaa67d293d71ba8cfdb8024" }
+tendermint = { git = "https://github.com/informalsystems/tendermint-rs.git", default-features = false, branch = "master" }
+tendermint-rpc = { git = "https://github.com/informalsystems/tendermint-rs.git", default-features = false, branch = "master" }
 tokio = { version = "0.2", features = ["rt-threaded", "sync", "time", "tcp"], optional = true }
 tokio-tungstenite = { version = "0.10", features = ["tls"], optional = true }
 uuid = { version = "0.8.1", features = ["v4"] }

client-common/src/cipher/mock.rs

@@ -48,10 +48,7 @@ where
             .collect::<Result<Vec<_>>>()
             .expect("abci_query failed");
 
-        let sealed_logs = rsps
-            .into_iter()
-            .map(|rsp| rsp.value.expect("sealed log query failed"))
-            .collect::<Vec<_>>();
+        let sealed_logs = rsps.into_iter().map(|rsp| rsp.value).collect::<Vec<_>>();
 
         let txs = sealed_logs
             .into_iter()
@@ -95,7 +92,6 @@ fn checked_unseal(payload: &[u8], _private_key: &PrivateKey) -> Option<TxWithOut
 mod tests {
     use super::*;
 
-    use crate::tendermint::lite;
     use crate::tendermint::types::*;
     use crate::PrivateKey;
     use chain_core::state::ChainState;
@@ -135,14 +131,6 @@ mod tests {
             unreachable!()
         }
 
-        fn block_batch_verified<'a, T: Clone + Iterator<Item = &'a u64>>(
-            &self,
-            _state: lite::TrustedState,
-            _heights: T,
-        ) -> Result<(Vec<Block>, lite::TrustedState)> {
-            unreachable!()
-        }
-
         fn broadcast_transaction(&self, _transaction: &[u8]) -> Result<BroadcastTxResponse> {
             unreachable!()
         }
@@ -155,7 +143,7 @@ mod tests {
             _prove: bool,
         ) -> Result<AbciQuery> {
             Ok(AbciQuery {
-                value: Some(seal(&TxWithOutputs::Transfer(Tx::default()))),
+                value: seal(&TxWithOutputs::Transfer(Tx::default())),
                 ..Default::default()
             })
         }

client-common/src/tendermint/client.rs

@@ -1,4 +1,3 @@
-use crate::tendermint::lite;
 use crate::tendermint::types::*;
 use crate::Result;
 use chain_core::state::ChainState;
@@ -26,13 +25,6 @@ pub trait Client: Send + Sync + Clone {
         heights: T,
     ) -> Result<Vec<BlockResultsResponse>>;
 
-    /// Fetch continuous blocks and verify them.
-    fn block_batch_verified<'a, T: Clone + Iterator<Item = &'a u64>>(
-        &self,
-        state: lite::TrustedState,
-        heights: T,
-    ) -> Result<(Vec<Block>, lite::TrustedState)>;
-
     /// Makes `broadcast_tx_sync` call to tendermint
     fn broadcast_transaction(&self, transaction: &[u8]) -> Result<BroadcastTxResponse>;
 

client-common/src/tendermint/rpc_client/async_rpc_client.rs

@@ -123,8 +123,11 @@ impl AsyncRpcClient {
             let method = batch_params[i].0;
             let params = &batch_params[i].1;
 
-            let response = self.receive_response(method, params, &id, receiver).await?;
-            responses.push(response);
+            if let Ok(response) = self.receive_response(method, params, &id, receiver).await {
+                responses.push(response);
+            } else {
+                break;
+            }
         }
 
         Ok(responses)
@@ -154,18 +157,12 @@ impl AsyncRpcClient {
         let response_values = self.request_batch(batch_params).await?;
         let mut responses = Vec::with_capacity(response_values.len());
 
-        for (i, response_value) in response_values.into_iter().enumerate() {
-            let method = batch_params[i].0;
-            let params = &batch_params[i].1;
-
-            let response = serde_json::from_value(response_value).with_context(|| {
-                format!(
-                    "Unable to deserialize `{}` from JSON-RPC response for params: {:?}",
-                    method, params
-                )
-            })?;
-
-            responses.push(response);
+        for response_value in response_values.into_iter() {
+            if let Ok(response) = serde_json::from_value(response_value) {
+                responses.push(response);
+            } else {
+                break;
+            }
         }
 
         Ok(responses)

client-common/src/tendermint/rpc_client/sync_rpc_client.rs

@@ -1,22 +1,20 @@
 use std::{
     convert::TryFrom,
     sync::{mpsc::sync_channel, Arc},
-    time::{Duration, SystemTime},
+    time::Duration,
 };
 
-use itertools::izip;
 use once_cell::sync::OnceCell;
 use serde::Deserialize;
 use serde_json::{json, Value};
-use tendermint::{lite, validator};
 use tokio::runtime::Runtime;
 
 use chain_core::state::ChainState;
 use std::sync::Mutex;
 
 use super::async_rpc_client::AsyncRpcClient;
 use crate::{
-    tendermint::{lite::TrustedState, types::*, Client},
+    tendermint::{types::*, Client},
     Error, ErrorKind, PrivateKey, Result, ResultExt, SignedTransaction, Transaction,
     TransactionObfuscation,
 };
@@ -200,31 +198,6 @@ impl SyncRpcClient {
                 )
             })
     }
-
-    fn validators_batch<T: Iterator<Item = u64>>(
-        &self,
-        heights: T,
-    ) -> Result<Vec<ValidatorsResponse>> {
-        let params = heights
-            .map(|height| {
-                (
-                    "validators",
-                    vec![json!(height.to_string()), json!("0"), json!("100")],
-                )
-            })
-            .collect::<Vec<(&str, Vec<Value>)>>();
-        self.call_batch(params)
-    }
-
-    fn commit_batch<'a, T: Iterator<Item = &'a u64>>(
-        &self,
-        heights: T,
-    ) -> Result<Vec<CommitResponse>> {
-        let params = heights
-            .map(|height| ("commit", vec![json!(height.to_string())]))
-            .collect::<Vec<(&str, Vec<Value>)>>();
-        self.call_batch(params)
-    }
 }
 
 impl Client for SyncRpcClient {
@@ -272,48 +245,6 @@ impl Client for SyncRpcClient {
         self.call_batch(params)
     }
 
-    /// Fetch continuous blocks and verify them.
-    fn block_batch_verified<'a, T: Clone + Iterator<Item = &'a u64>>(
-        &self,
-        mut state: TrustedState,
-        heights: T,
-    ) -> Result<(Vec<Block>, TrustedState)> {
-        let commits = self.commit_batch(heights.clone())?;
-        let validators: Vec<validator::Set> = self
-            .validators_batch(heights.clone().map(|h| h.saturating_add(1)))?
-            .into_iter()
-            .map(|rsp| validator::Set::new(rsp.validators))
-            .collect();
-        let blocks = self.block_batch(heights)?;
-        for (commit, next_vals, block) in izip!(&commits, &validators, &blocks) {
-            let signed_header =
-                lite::SignedHeader::new(commit.signed_header.clone(), block.header.clone());
-            state = if let Some(state) = &state.0 {
-                lite::verifier::verify_single(
-                    state.clone(),
-                    &signed_header,
-                    state.validators(),
-                    next_vals,
-                    // FIXME make parameters configurable
-                    lite::TrustThresholdFraction::new(1, 3).unwrap(),
-                    Duration::from_secs(std::u32::MAX as u64),
-                    SystemTime::now(),
-                )
-                .map_err(|err| {
-                    Error::new(
-                        ErrorKind::VerifyError,
-                        format!("block verify failed: {:?}", err),
-                    )
-                })?
-                .into()
-            } else {
-                // TODO verify block1 against genesis block
-                lite::TrustedState::new(signed_header, next_vals.clone()).into()
-            };
-        }
-        Ok((blocks, state))
-    }
-
     /// Makes `broadcast_tx_sync` call to tendermint
     fn broadcast_transaction(&self, transaction: &[u8]) -> Result<BroadcastTxResponse> {
         let params = vec![json!(transaction)];
@@ -378,23 +309,12 @@ impl Client for SyncRpcClient {
 
         rsps.into_iter()
             .map(|rsp| {
-                if let Some(value) = rsp.response.value {
-                    let state = serde_json::from_str(
-                        &String::from_utf8(value)
-                            .chain(|| (ErrorKind::InvalidInput, "chain state decode failed"))?,
-                    )
+                let state = serde_json::from_str(
+                    &String::from_utf8(rsp.response.value)
+                    .chain(|| (ErrorKind::InvalidInput, "chain state decode failed"))?,
+                )
                     .chain(|| (ErrorKind::InvalidInput, "chain state decode failed"))?;
-                    Ok(state)
-                } else {
-                    Err(Error::new(
-                        ErrorKind::InvalidInput,
-                        format!(
-                            "abci query fail: {}, {}",
-                            rsp.response.code.value(),
-                            rsp.response.log,
-                        ),
-                    ))
-                }
+                Ok(state)
             })
             .collect()
     }

client-common/src/tendermint/types.rs

@@ -11,17 +11,16 @@ use chain_core::tx::fee::LinearFee;
 use chain_core::tx::{TxAux, TxEnclaveAux, TxPublicAux};
 
 pub use self::block_results::BlockResults;
-pub use tendermint::rpc::endpoint::{
-    abci_query::AbciQuery, abci_query::Response as AbciQueryResponse,
-    block::Response as BlockResponse, block_results::Response as BlockResultsResponse,
-    broadcast::tx_sync::Response as BroadcastTxResponse, commit::Response as CommitResponse,
-    status::Response as StatusResponse, validators::Response as ValidatorsResponse,
-};
-pub use tendermint::rpc::endpoint::{broadcast, status};
 pub use tendermint::{
     abci, abci::transaction::Data, abci::Code, block::Header, block::Height, Block,
     Genesis as GenericGenesis, Hash, Time,
 };
+pub use tendermint_rpc::endpoint::{
+    abci_query::AbciQuery, abci_query::Response as AbciQueryResponse,
+    block::Response as BlockResponse, block_results::Response as BlockResultsResponse, broadcast,
+    broadcast::tx_sync::Response as BroadcastTxResponse, commit::Response as CommitResponse,
+    status, status::Response as StatusResponse, validators::Response as ValidatorsResponse,
+};
 
 /// crypto-com instantiated genesis type
 pub type Genesis = GenericGenesis<Option<InitConfig>>;
@@ -118,6 +117,6 @@ pub trait AbciQueryExt {
 
 impl AbciQueryExt for AbciQuery {
     fn bytes(&self) -> Vec<u8> {
-        self.value.clone().unwrap_or_default()
+        self.value.clone()
     }
 }