Chain Runtime implementation (informalsystems#364)

* Started v0 impl: re-arranging handlers + initialization. * Clean up chain handle, prepare for generic client creation. * Trying different boundaries of abstraction b/t ForeignClient and ChainHandle. * Version one: introduced assemble_ methods to construct chain-specific data types * Removed signer from foreign client config * FMT quick fix * Client ID parsing * Use ibc::ics24_host::identifier::ChainId instead of its tendermint counterpart * More error handling and split into modules * Move prod handle in its own module * Cleanup * Overhaul subscriptions handling * Add struct to represent a batch of events * Move SplitResults trait into its own module * WIP: More handle inputs * Better errors * Implement `query` in chain runtime * Assemble client state and consensus state in runtime * Launch light clients * Light Block * Remove unused file * Rename two handle inputs * Rename Chain::ics_query to Chain::query * Store the domain MerkleProof in QueryResponse instead of the raw MerkleProof * Implement all chain methods in runtime + handle * Chain methods should return their Self types * Use light client to build consensus state * Use light client to retrieve and build header * Comment out currently unused code * Formatting * Make keybase field of CosmosSDKChain private * Fix tests * Properly spawn runtime * Revert 2afa90f * Allow a ChainId to not include version number, default to 0 * Upgrade Tokio to version 0.3 * WIP: event monitor * Syncify EventMonitor * Cleanup * Use shared Tokio runtime to `block_on` instead of spawning new one * Cleanup * Nest event_handler and event_monitor modules under event module * WIP: event bus * Cleanup * Fix build issue * Add tests for event bus * Cleanup * Update listen command * Point to tendermint-rs branch with Tokio 0.3 * Remove unused downcast_* methods * Re-enable some lints * Cleanup * Use tendermint-rs master * Depend on tendermint-rs prost-dev branch * Formatting * Deserialize client queries in Any types, temp fix for validator set Co-authored-by: Adi Seredinschi <[email protected]> Co-authored-by: Anca Zamfir <[email protected]>
hu55a1n1 · Nov 23, 2020 · d2cbc2b · d2cbc2b
1 parent d846a6c
commit d2cbc2b
Show file tree

Hide file tree

Showing 55 changed files with 2,571 additions and 1,615 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -9,4 +9,19 @@ members = [
 
 exclude = [
     "proto-compiler"
-]
+]
+
+[patch.crates-io]
+tendermint              = { git = "https://github.com/informalsystems/tendermint-rs", branch = "romac/prost-dev" }
+tendermint-rpc          = { git = "https://github.com/informalsystems/tendermint-rs", branch = "romac/prost-dev" }
+tendermint-proto        = { git = "https://github.com/informalsystems/tendermint-rs", branch = "romac/prost-dev" }
+tendermint-light-client = { git = "https://github.com/informalsystems/tendermint-rs", branch = "romac/prost-dev" }
+
+bytes = { git = "https://github.com/tokio-rs/bytes", tag = "v0.6.0" }
+
+tonic = { git = "https://github.com/romac/tonic", branch = "romac/hyper-0.14-dev" }
+
+prost        = { git = "https://github.com/danburkert/prost", branch = "master" }
+prost-types  = { git = "https://github.com/danburkert/prost", branch = "master" }
+prost-build  = { git = "https://github.com/danburkert/prost", branch = "master" }
+prost-derive = { git = "https://github.com/danburkert/prost", branch = "master" }
diff --git a/relayer-cli/Cargo.toml b/relayer-cli/Cargo.toml
@@ -23,6 +23,7 @@ sled = "0.34.4"
 prost = "0.6.1"
 prost-types = { version = "0.6.1" }
 hex = "0.4"
+crossbeam-channel = "0.5.0"
 
 [dependencies.tendermint-proto]
 version = "=0.17.0-rc3"

diff --git a/relayer-cli/src/commands/light/add.rs b/relayer-cli/src/commands/light/add.rs
@@ -1,12 +1,14 @@
-use std::{fmt, io, io::Write, ops::Deref};
+use std::{fmt, io, io::Write, ops::Deref, path::PathBuf};
 
 use abscissa_core::{application::fatal_error, error::BoxError, Command, Options, Runnable};
 
+use config::StoreConfig;
+use ibc::ics24_host::identifier::ChainId;
 use relayer::{
+    config,
     config::{Config, LightClientConfig, PeersConfig},
     util::block_on,
 };
-use tendermint::chain::Id as ChainId;
 use tendermint::hash::Hash;
 use tendermint::{block::Height, net};
 use tendermint_light_client::types::PeerId;
@@ -24,6 +26,9 @@ pub struct AddCmd {
     #[options(short = "c")]
     chain_id: Option<ChainId>,
 
+    /// Path to light client store for this peer
+    store_path: Option<PathBuf>,
+
     /// whether this is the primary peer
     primary: bool,
 
@@ -42,6 +47,9 @@ struct AddOptions {
     /// RPC network address
     address: net::Address,
 
+    /// Path to light client store for this peer
+    store_path: PathBuf,
+
     /// whether this is the primary peer or not
     primary: bool,
 
@@ -56,18 +64,38 @@ impl AddOptions {
     fn from_cmd(cmd: &AddCmd) -> Result<AddOptions, BoxError> {
         let chain_id = cmd.chain_id.clone().ok_or("missing chain identifier")?;
         let address = cmd.address.clone().ok_or("missing RPC network address")?;
+        let store_path = cmd.store_path.clone().ok_or("missing store path")?;
         let primary = cmd.primary;
         let force = cmd.force;
         let yes = cmd.yes;
 
         Ok(AddOptions {
             chain_id,
             address,
+            store_path,
             primary,
             force,
             yes,
         })
     }
+
+    fn validate(&self) -> Result<(), BoxError> {
+        if !self.store_path.exists() {
+            return Err(
+                format!("Store path '{}' does not exists", self.store_path.display()).into(),
+            );
+        }
+
+        if !self.store_path.is_dir() {
+            return Err(format!(
+                "Store path '{}' is not a directory",
+                self.store_path.display()
+            )
+            .into());
+        }
+
+        Ok(())
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -180,8 +208,12 @@ fn update_config(
     let light_client_config = LightClientConfig {
         peer_id: status.peer_id,
         address: status.address.clone(),
+        timeout: config::default::timeout(),
         trusted_header_hash: status.latest_hash,
         trusted_height: status.latest_height,
+        store: StoreConfig::Disk {
+            path: options.store_path.join(status.peer_id.to_string()),
+        },
     };
 
     if peer_exists {
@@ -205,6 +237,10 @@ impl AddCmd {
         let config = (*app_config()).clone();
         let options = AddOptions::from_cmd(self).map_err(|e| format!("invalid options: {}", e))?;
 
+        options
+            .validate()
+            .map_err(|e| format!("invalid options: {}", e))?;
+
         add(config, options)
     }
 }

diff --git a/relayer-cli/src/commands/light/rm.rs b/relayer-cli/src/commands/light/rm.rs
@@ -4,8 +4,8 @@ use crate::prelude::*;
 
 use abscissa_core::{application::fatal_error, error::BoxError, Command, Options, Runnable};
 
+use ibc::ics24_host::identifier::ChainId;
 use relayer::config::Config;
-use tendermint::chain::Id as ChainId;
 use tendermint_light_client::types::PeerId;
 
 #[derive(Command, Debug, Options)]

diff --git a/relayer-cli/src/commands/listen.rs b/relayer-cli/src/commands/listen.rs
@@ -1,31 +1,67 @@
-use std::ops::Deref;
+use std::{ops::Deref, sync::Arc, thread};
 
-use abscissa_core::{
-    application::fatal_error, error::BoxError, tracing::debug, Command, Options, Runnable,
-};
+use abscissa_core::{application::fatal_error, error::BoxError, Command, Options, Runnable};
+use crossbeam_channel as channel;
+use tokio::runtime::Runtime as TokioRuntime;
 
-use crate::{prelude::*, tasks::event_listener};
+use ibc::ics24_host::identifier::ChainId;
+use relayer::{config::ChainConfig, event::monitor::*};
+
+use crate::prelude::*;
 
 #[derive(Command, Debug, Options)]
-pub struct ListenCmd {}
+pub struct ListenCmd {
+    #[options(free)]
+    chain_id: Option<ChainId>,
+}
 
 impl ListenCmd {
-    async fn cmd(&self) -> Result<(), BoxError> {
+    fn cmd(&self) -> Result<(), BoxError> {
+        let rt = Arc::new(TokioRuntime::new().unwrap());
         let config = app_config().clone();
 
-        debug!("launching 'listen' command");
-        event_listener::start(&config, false).await
+        let chain_id = self.chain_id.clone().unwrap();
+        let chain_config = config
+            .chains
+            .into_iter()
+            .find(|c| c.id == chain_id)
+            .unwrap();
+
+        listen(rt, chain_config)
     }
 }
 
 impl Runnable for ListenCmd {
     fn run(&self) {
-        let rt = tokio::runtime::Runtime::new().unwrap();
+        self.cmd()
+            .unwrap_or_else(|e| fatal_error(app_reader().deref(), &*e));
+    }
+}
+
+/// Listen to events
+pub fn listen(rt: Arc<TokioRuntime>, config: ChainConfig) -> Result<(), BoxError> {
+    info!(chain.id = %config.id, "spawning event monitor for");
 
-        rt.block_on(async move {
-            self.cmd()
-                .await
-                .unwrap_or_else(|e| fatal_error(app_reader().deref(), &*e));
-        });
+    let (event_monitor, rx) = subscribe(config, rt)?;
+    let _ = thread::spawn(|| event_monitor.run());
+
+    while let Ok(event_batch) = rx.recv() {
+        dbg!(event_batch);
     }
+
+    Ok(())
+}
+
+fn subscribe(
+    chain_config: ChainConfig,
+    rt: Arc<tokio::runtime::Runtime>,
+) -> Result<(EventMonitor, channel::Receiver<EventBatch>), BoxError> {
+    let (mut event_monitor, rx) = EventMonitor::new(chain_config.id, chain_config.rpc_addr, rt)
+        .map_err(|e| format!("couldn't initialize event monitor: {}", e))?;
+
+    event_monitor
+        .subscribe()
+        .map_err(|e| format!("couldn't initialize subscriptions: {}", e))?;
+
+    Ok((event_monitor, rx))
 }
diff --git a/relayer-cli/src/commands/query/channel.rs b/relayer-cli/src/commands/query/channel.rs
@@ -1,19 +1,22 @@
-use crate::prelude::*;
-use std::convert::TryInto;
+use std::sync::Arc;
 
 use abscissa_core::{Command, Options, Runnable};
-use relayer::config::{ChainConfig, Config};
+use tokio::runtime::Runtime as TokioRuntime;
 
 use ibc::ics04_channel::channel::ChannelEnd;
+use ibc::ics24_host::error::ValidationError;
+use ibc::ics24_host::identifier::ChainId;
 use ibc::ics24_host::identifier::{ChannelId, PortId};
 use ibc::ics24_host::Path::ChannelEnds;
 
-use crate::error::{Error, Kind};
-use ibc::ics24_host::error::ValidationError;
 use relayer::chain::{Chain, CosmosSDKChain};
-use tendermint::chain::Id as ChainId;
+use relayer::config::{ChainConfig, Config};
+
 use tendermint_proto::Protobuf;
 
+use crate::error::{Error, Kind};
+use crate::prelude::*;
+
 #[derive(Clone, Command, Debug, Options)]
 pub struct QueryChannelEndCmd {
     #[options(free, help = "identifier of the chain to query")]
@@ -94,11 +97,15 @@ impl Runnable for QueryChannelEndCmd {
 
         // run without proof:
         // cargo run --bin relayer -- -c relayer/tests/config/fixtures/simple_config.toml query channel end ibc-test firstport firstchannel --height 3 -p false
-        let chain = CosmosSDKChain::from_config(chain_config).unwrap();
+
+        let rt = Arc::new(TokioRuntime::new().unwrap());
+
+        let chain = CosmosSDKChain::from_config(chain_config, rt).unwrap();
+        let height = ibc::Height::new(chain.id().version(), opts.height);
         let res: Result<ChannelEnd, Error> = chain
             .query(
                 ChannelEnds(opts.port_id, opts.channel_id),
-                opts.height.try_into().unwrap(),
+                height,
                 opts.proof,
             )
             .map_err(|e| Kind::Query.context(e).into())

diff --git a/relayer-cli/src/commands/query/client.rs b/relayer-cli/src/commands/query/client.rs
@@ -1,20 +1,23 @@
-use crate::prelude::*;
+use std::sync::Arc;
 
 use abscissa_core::{Command, Options, Runnable};
-use relayer::config::{ChainConfig, Config};
+use tokio::runtime::Runtime as TokioRuntime;
 
-use crate::error::{Error, Kind};
 use ibc::ics02_client::client_def::{AnyClientState, AnyConsensusState};
 use ibc::ics02_client::raw::ConnectionIds as ConnectionIDs;
 use ibc::ics24_host::error::ValidationError;
+use ibc::ics24_host::identifier::ChainId;
 use ibc::ics24_host::identifier::ClientId;
 use ibc::ics24_host::Path::{ClientConnections, ClientConsensusState, ClientState};
+
+use tendermint_proto::Protobuf;
+
 use relayer::chain::Chain;
 use relayer::chain::CosmosSDKChain;
-use tendermint::chain::Id as ChainId;
-use tendermint_proto::Protobuf;
+use relayer::config::{ChainConfig, Config};
 
-use std::convert::TryInto;
+use crate::error::{Error, Kind};
+use crate::prelude::*;
 
 /// Query client state command
 #[derive(Clone, Command, Debug, Options)]
@@ -75,14 +78,12 @@ impl Runnable for QueryClientStateCmd {
         };
         status_info!("Options", "{:?}", opts);
 
-        let chain = CosmosSDKChain::from_config(chain_config).unwrap();
+        let rt = Arc::new(TokioRuntime::new().unwrap());
+        let chain = CosmosSDKChain::from_config(chain_config, rt).unwrap();
+        let height = ibc::Height::new(chain.id().version(), opts.height);
 
         let res: Result<AnyClientState, Error> = chain
-            .query(
-                ClientState(opts.client_id),
-                opts.height.try_into().unwrap(),
-                opts.proof,
-            )
+            .query(ClientState(opts.client_id), height, opts.proof)
             .map_err(|e| Kind::Query.context(e).into())
             .and_then(|v| {
                 AnyClientState::decode_vec(&v.value).map_err(|e| Kind::Query.context(e).into())
@@ -170,15 +171,18 @@ impl Runnable for QueryClientConsensusCmd {
         };
         status_info!("Options", "{:?}", opts);
 
-        let chain = CosmosSDKChain::from_config(chain_config).unwrap();
+        let rt = Arc::new(TokioRuntime::new().unwrap());
+        let chain = CosmosSDKChain::from_config(chain_config, rt).unwrap();
+        let height = ibc::Height::new(chain.id().version(), opts.height);
+
         let res: Result<AnyConsensusState, Error> = chain
             .query(
                 ClientConsensusState {
                     client_id: opts.client_id,
                     epoch: opts.version_number,
                     height: opts.version_height,
                 },
-                opts.height.try_into().unwrap(),
+                height,
                 opts.proof,
             )
             .map_err(|e| Kind::Query.context(e).into())
@@ -281,17 +285,17 @@ impl Runnable for QueryClientConnectionsCmd {
         };
         status_info!("Options", "{:?}", opts);
 
-        let chain = CosmosSDKChain::from_config(chain_config).unwrap();
+        let rt = Arc::new(TokioRuntime::new().unwrap());
+        let chain = CosmosSDKChain::from_config(chain_config, rt).unwrap();
+        let height = ibc::Height::new(chain.id().version(), opts.height);
+
         let res: Result<ConnectionIDs, Error> = chain
-            .query(
-                ClientConnections(opts.client_id),
-                opts.height.try_into().unwrap(),
-                false,
-            )
+            .query(ClientConnections(opts.client_id), height, false)
             .map_err(|e| Kind::Query.context(e).into())
             .and_then(|v| {
                 ConnectionIDs::decode_vec(&v.value).map_err(|e| Kind::Query.context(e).into())
             });
+
         match res {
             Ok(cs) => status_info!("client connections query result: ", "{:?}", cs),
             Err(e) => status_info!("client connections query error", "{}", e),