feat: rpc as a task with publishing signature (#132)

* Renamed contract_updater to rpc

* Move fetch state and config to RpcClient

* Moved rpc functions over to rpc_client

* Made signature publishing be a part of rpc task

* Make EthClient and use rpc task for eth publish

* Some cleanup

* Address comments

chain-signatures/node/src/cli.rs

@@ -4,6 +4,7 @@ use crate::mesh::Mesh;
 use crate::node_client::{self, NodeClient};
 use crate::protocol::message::MessageChannel;
 use crate::protocol::{MpcSignProtocol, SignQueue};
+use crate::rpc::{EthClient, NearClient, RpcExecutor};
 use crate::storage::app_data_storage;
 use crate::{indexer, indexer_eth, logs, mesh, storage, web};
 use clap::Parser;
@@ -230,19 +231,30 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
             let client = NodeClient::new(&message_options);
             let signer = InMemorySigner::from_secret_key(account_id.clone(), account_sk);
             let (mesh, mesh_state) = Mesh::init(&client, mesh_options);
+            let contract_state = Arc::new(RwLock::new(None));
+
+            let network = NetworkConfig {
+                cipher_sk: hpke::SecretKey::try_from_bytes(&hex::decode(cipher_sk)?)?,
+                cipher_pk: hpke::PublicKey::try_from_bytes(&hex::decode(cipher_pk)?)?,
+                sign_sk,
+            };
+            let eth_client = EthClient::new(&indexer_eth_options, &eth_account_sk);
+            let near_client =
+                NearClient::new(&near_rpc, &my_address, &network, &mpc_contract_id, signer);
+            let (rpc_channel, rpc) = RpcExecutor::new(&near_client, &eth_client);
             let config = Arc::new(RwLock::new(Config::new(LocalConfig {
                 over: override_config.unwrap_or_else(Default::default),
-                network: NetworkConfig {
-                    cipher_sk: hpke::SecretKey::try_from_bytes(&hex::decode(cipher_sk)?)?,
-                    cipher_pk: hpke::PublicKey::try_from_bytes(&hex::decode(cipher_pk)?)?,
-                    sign_sk,
-                },
+                network,
             })));
-            let contract_state = Arc::new(RwLock::new(None));
-
-            let contract_updater =
-                crate::contract_updater::ContractUpdater::init(&rpc_client, &mpc_contract_id);
 
+            tracing::info!(
+                ?mpc_contract_id,
+                ?account_id,
+                ?my_address,
+                near_rpc_url = ?near_client.rpc_addr(),
+                eth_rpc_url = ?indexer_eth_options.eth_rpc_http_url,
+                "starting node",
+            );
             rt.block_on(async {
                 let state = Arc::new(RwLock::new(crate::protocol::NodeState::Starting));
                 let (sender, channel) =
@@ -252,21 +264,17 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
                     mpc_contract_id,
                     account_id.clone(),
                     state.clone(),
-                    rpc_client,
-                    signer,
+                    near_client,
+                    rpc_channel,
                     channel,
                     sign_rx,
                     key_storage,
                     triple_storage,
                     presignature_storage,
-                    indexer_eth_options.eth_rpc_http_url.clone(),
-                    indexer_eth_options.eth_contract_address.clone(),
-                    eth_account_sk,
                 );
 
                 tracing::info!("protocol initialized");
-                let contract_handle =
-                    tokio::spawn(contract_updater.run(contract_state.clone(), config.clone()));
+                let rpc_handle = tokio::spawn(rpc.run(contract_state.clone(), config.clone()));
                 let mesh_handle = tokio::spawn(mesh.run(contract_state.clone()));
                 let protocol_handle =
                     tokio::spawn(protocol.run(contract_state, config, mesh_state));
@@ -276,7 +284,7 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
                     tokio::spawn(indexer_eth::run(indexer_eth_options, sign_tx, account_id));
                 tracing::info!("protocol http server spawned");
 
-                contract_handle.await??;
+                rpc_handle.await?;
                 mesh_handle.await??;
                 protocol_handle.await??;
                 web_handle.await??;

chain-signatures/node/src/config.rs

@@ -3,10 +3,11 @@ use std::str::FromStr;
 
 use mpc_contract::config::ProtocolConfig;
 use mpc_keys::hpke;
-use near_account_id::AccountId;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
+use crate::rpc::NearClient;
+
 /// The contract's config is a dynamic representation of all configurations possible.
 pub type ContractConfig = HashMap<String, Value>;
 
@@ -51,12 +52,8 @@ impl Config {
 
     /// Fetches the latest config from the contract and set the config inplace. The old config
     /// is returned when swap is completed.
-    pub async fn fetch_inplace(
-        &mut self,
-        rpc_client: &near_fetch::Client,
-        contract_id: &AccountId,
-    ) -> anyhow::Result<Self> {
-        let new_config = crate::rpc_client::fetch_mpc_config(rpc_client, contract_id, self).await?;
+    pub async fn fetch_inplace(&mut self, rpc_client: &NearClient) -> anyhow::Result<Self> {
+        let new_config = rpc_client.fetch_config(self).await?;
         Ok(std::mem::replace(self, new_config))
     }
 }

chain-signatures/node/src/lib.rs

@@ -1,6 +1,5 @@
 pub mod cli;
 pub mod config;
-pub mod contract_updater;
 pub mod gcp;
 pub mod indexer;
 pub mod indexer_eth;
@@ -10,7 +9,7 @@ pub mod mesh;
 pub mod metrics;
 pub mod node_client;
 pub mod protocol;
-pub mod rpc_client;
+pub mod rpc;
 pub mod storage;
 pub mod types;
 pub mod util;

chain-signatures/node/src/protocol/consensus.rs

@@ -11,7 +11,7 @@ use crate::protocol::signature::SignatureManager;
 use crate::protocol::state::{GeneratingState, ResharingState};
 use crate::protocol::triple::TripleManager;
 use crate::protocol::SignRequest;
-use crate::rpc_client;
+use crate::rpc::NearClient;
 use crate::storage::presignature_storage::PresignatureStorage;
 use crate::storage::secret_storage::SecretNodeStorageBox;
 use crate::storage::triple_storage::TripleStorage;
@@ -23,17 +23,14 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use cait_sith::protocol::InitializationError;
-use serde_json::json;
 use tokio::sync::{mpsc, RwLock};
 use url::Url;
 
 use near_account_id::AccountId;
-use near_crypto::InMemorySigner;
 
 pub trait ConsensusCtx {
     fn my_account_id(&self) -> &AccountId;
-    fn rpc_client(&self) -> &near_fetch::Client;
-    fn signer(&self) -> &InMemorySigner;
+    fn near_client(&self) -> &NearClient;
     fn mpc_contract_id(&self) -> &AccountId;
     fn my_address(&self) -> &Url;
     fn sign_rx(&self) -> Arc<RwLock<mpsc::Receiver<SignRequest>>>;
@@ -298,14 +295,10 @@ impl ConsensusProtocol for WaitingForConsensusState {
                     .unwrap_or_default();
                 if !has_voted {
                     tracing::info!("waiting(initializing): we haven't voted yet, voting for the generated public key");
-                    rpc_client::vote_for_public_key(
-                        ctx.rpc_client(),
-                        ctx.signer(),
-                        ctx.mpc_contract_id(),
-                        &public_key,
-                    )
-                    .await
-                    .map_err(|err| ConsensusError::CannotVote(format!("{err:?}")))?;
+                    ctx.near_client()
+                        .vote_public_key(&public_key)
+                        .await
+                        .map_err(|err| ConsensusError::CannotVote(format!("{err:?}")))?;
                 }
                 Ok(NodeState::WaitingForConsensus(self))
             }
@@ -439,16 +432,10 @@ impl ConsensusProtocol for WaitingForConsensusState {
                                         epoch = self.epoch,
                                         "waiting(resharing): we haven't voted yet, voting for resharing to complete"
                                     );
-                                    rpc_client::vote_reshared(
-                                        ctx.rpc_client(),
-                                        ctx.signer(),
-                                        ctx.mpc_contract_id(),
-                                        self.epoch,
-                                    )
-                                    .await
-                                    .map_err(|err| {
-                                        ConsensusError::CannotVote(format!("{err:?}"))
-                                    })?;
+
+                                    ctx.near_client().vote_reshared(self.epoch).await.map_err(
+                                        |err| ConsensusError::CannotVote(format!("{err:?}")),
+                                    )?;
                                 } else {
                                     tracing::info!(
                                         epoch = self.epoch,
@@ -626,7 +613,7 @@ impl ConsensusProtocol for JoiningState {
         self,
         ctx: C,
         contract_state: ProtocolState,
-        cfg: Config,
+        _cfg: Config,
     ) -> Result<NodeState, ConsensusError> {
         match contract_state {
             ProtocolState::Initializing(_) => Err(ConsensusError::ContractStateRollback),
@@ -659,21 +646,10 @@ impl ConsensusProtocol for JoiningState {
                         tracing::info!(
                             "joining(running): sending a transaction to join the participant set"
                         );
-                        ctx.rpc_client()
-                            .call(ctx.signer(), ctx.mpc_contract_id(), "join")
-                            .args_json(json!({
-                                "url": ctx.my_address(),
-                                "cipher_pk": cfg.local.network.cipher_pk.to_bytes(),
-                                "sign_pk": cfg.local.network.sign_sk.public_key(),
-                            }))
-                            .max_gas()
-                            .retry_exponential(10, 3)
-                            .transact()
-                            .await
-                            .map_err(|err| {
-                                tracing::error!(?err, "failed to join the participant set");
-                                ConsensusError::CannotJoin(format!("{err:?}"))
-                            })?;
+                        ctx.near_client().propose_join().await.map_err(|err| {
+                            tracing::error!(?err, "failed to join the participant set");
+                            ConsensusError::CannotJoin(format!("{err:?}"))
+                        })?;
                         Ok(NodeState::Joining(self))
                     }
                 }

chain-signatures/node/src/protocol/cryptography.rs

@@ -7,25 +7,20 @@ use crate::protocol::message::{GeneratingMessage, ResharingMessage};
 use crate::protocol::presignature::PresignatureManager;
 use crate::protocol::state::{PersistentNodeData, WaitingForConsensusState};
 use crate::protocol::MeshState;
+use crate::rpc::RpcChannel;
 use crate::storage::secret_storage::SecretNodeStorageBox;
 
 use async_trait::async_trait;
 use cait_sith::protocol::{Action, InitializationError, ProtocolError};
 use k256::elliptic_curve::group::GroupEncoding;
 use near_account_id::AccountId;
-use near_crypto::InMemorySigner;
-use web3::Web3;
 
 pub trait CryptographicCtx {
-    fn rpc_client(&self) -> &near_fetch::Client;
-    fn eth_client(&self) -> &Web3<web3::transports::Http>;
-    fn eth_contract_address(&self) -> String;
-    fn eth_account_sk(&self) -> String;
-    fn signer(&self) -> &InMemorySigner;
     fn mpc_contract_id(&self) -> &AccountId;
     fn secret_storage(&mut self) -> &mut SecretNodeStorageBox;
     fn my_account_id(&self) -> &AccountId;
     fn channel(&self) -> &MessageChannel;
+    fn rpc_channel(&self) -> &RpcChannel;
 }
 
 #[derive(thiserror::Error, Debug)]