remove repeated code (#258)

Cargo.toml

@@ -21,6 +21,7 @@ members = [
   "chains/polygon/rosetta-testing-polygon",
   "chains/binance",
   "chains/avalanche",
+  "chains/rosetta-chain-testing",
 ]
 resolver = "2"
 

chains/avalanche/Cargo.toml

@@ -13,6 +13,7 @@ ethers = { version = "2.0", default-features = true, features = ["abigen", "rust
 ethers-solc = "2.0"
 hex-literal = "0.4"
 rand_core = { version = "0.6", features = ["getrandom"] }
+rosetta-chain-testing = { path = "../rosetta-chain-testing" }
 rosetta-client.workspace = true
 rosetta-config-ethereum.workspace = true
 rosetta-core.workspace = true

chains/avalanche/src/lib.rs

@@ -44,13 +44,14 @@ mod tests {
     use ethers::types::H256;
     use ethers_solc::{artifacts::Source, CompilerInput, EvmVersion, Solc};
     use hex_literal::hex;
+    use rosetta_chain_testing::run_test;
     use rosetta_client::Wallet;
     use rosetta_config_ethereum::{AtBlock, CallResult};
     use rosetta_core::BlockchainClient;
     use rosetta_server_ethereum::MaybeWsEthereumClient;
     use serial_test::serial;
     use sha3::Digest;
-    use std::{collections::BTreeMap, future::Future, path::Path};
+    use std::{collections::BTreeMap, path::Path};
 
     /// Account used to fund other testing accounts.
     const FUNDING_ACCOUNT_PRIVATE_KEY: [u8; 32] =
@@ -68,38 +69,6 @@ mod tests {
         }
     }
 
-    /// Run the test in another thread while sending txs to force binance to mine new blocks
-    /// # Panic
-    /// Panics if the future panics
-    async fn run_test<Fut: Future<Output = ()> + Send + 'static>(future: Fut) {
-        // Guarantee that only one test is incrementing blocks at a time
-        static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
-
-        // Run the test in another thread
-        let test_handler = tokio::spawn(future);
-
-        // Acquire Lock
-        let guard = LOCK.lock().await;
-
-        // Check if the test is finished after acquiring the lock
-        if test_handler.is_finished() {
-            // Release lock
-            drop(guard);
-
-            // Now is safe to panic
-            if let Err(err) = test_handler.await {
-                std::panic::resume_unwind(err.into_panic());
-            }
-            return;
-        }
-
-        // Now is safe to panic
-        if let Err(err) = test_handler.await {
-            // Resume the panic on the main task
-            std::panic::resume_unwind(err.into_panic());
-        }
-    }
-
     #[tokio::test]
     #[serial]
     async fn network_status() {

chains/binance/Cargo.toml

@@ -12,6 +12,7 @@ anyhow = "1.0"
 ethers = { version = "2.0", default-features = true, features = ["abigen", "rustls", "ws"] }
 ethers-solc = "2.0"
 hex-literal = "0.4"
+rosetta-chain-testing = { path = "../rosetta-chain-testing" }
 rosetta-client.workspace = true
 rosetta-config-ethereum.workspace = true
 rosetta-core.workspace = true

chains/binance/src/lib.rs

@@ -43,12 +43,13 @@ mod tests {
 
     use ethers_solc::{artifacts::Source, CompilerInput, EvmVersion, Solc};
     use hex_literal::hex;
+    use rosetta_chain_testing::run_test;
     use rosetta_client::Wallet;
     use rosetta_config_ethereum::{AtBlock, CallResult};
     use rosetta_core::BlockchainClient;
     use rosetta_server_ethereum::MaybeWsEthereumClient;
     use sha3::Digest;
-    use std::{collections::BTreeMap, future::Future, path::Path};
+    use std::{collections::BTreeMap, path::Path};
 
     /// Binance rpc url
     const BINANCE_RPC_WS_URL: &str = "ws://127.0.0.1:8546";
@@ -62,38 +63,6 @@ mod tests {
         }
     }
 
-    /// Run the test in another thread while sending txs to force binance to mine new blocks
-    /// # Panic
-    /// Panics if the future panics
-    async fn run_test<Fut: Future<Output = ()> + Send + 'static>(future: Fut) {
-        // Guarantee that only one test is incrementing blocks at a time
-        static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
-
-        // Run the test in another thread
-        let test_handler: tokio::task::JoinHandle<()> = tokio::spawn(future);
-
-        // Acquire Lock
-        let guard = LOCK.lock().await;
-
-        // Check if the test is finished after acquiring the lock
-        if test_handler.is_finished() {
-            // Release lock
-            drop(guard);
-
-            // Now is safe to panic
-            if let Err(err) = test_handler.await {
-                std::panic::resume_unwind(err.into_panic());
-            }
-            return;
-        }
-
-        // Now is safe to panic
-        if let Err(err) = test_handler.await {
-            // Resume the panic on the main task
-            std::panic::resume_unwind(err.into_panic());
-        }
-    }
-
     #[tokio::test]
     async fn network_status() {
         run_test(async move {

chains/ethereum/backend/src/jsonrpsee.rs

@@ -398,8 +398,14 @@ where
     T: SubscriptionClientT + Send + Sync,
 {
     type SubscriptionError = serde_json::Error;
-    type NewHeadsStream<'a> = Subscription<RpcBlock<H256>> where Self: 'a;
-    type LogsStream<'a> = Subscription<Log> where Self: 'a;
+    type NewHeadsStream<'a>
+        = Subscription<RpcBlock<H256>>
+    where
+        Self: 'a;
+    type LogsStream<'a>
+        = Subscription<Log>
+    where
+        Self: 'a;
 
     /// Fires a notification each time a new header is appended to the chain, including chain
     /// reorganizations.

chains/ethereum/backend/src/lib.rs

@@ -313,8 +313,14 @@ pub trait EthereumPubSub: EthereumRpc {
 
 impl<'b, T: 'b + EthereumPubSub + ?Sized> EthereumPubSub for &'b T {
     type SubscriptionError = T::SubscriptionError;
-    type NewHeadsStream<'a> = T::NewHeadsStream<'a> where Self: 'a;
-    type LogsStream<'a> = T::LogsStream<'a> where Self: 'a;
+    type NewHeadsStream<'a>
+        = T::NewHeadsStream<'a>
+    where
+        Self: 'a;
+    type LogsStream<'a>
+        = T::LogsStream<'a>
+    where
+        Self: 'a;
     fn new_heads<'a, 'async_trait>(
         &'a self,
     ) -> BoxFuture<'async_trait, Result<Self::NewHeadsStream<'a>, Self::Error>>
@@ -342,8 +348,14 @@ impl<'b, T: 'b + EthereumPubSub + ?Sized> EthereumPubSub for &'b T {
 // https://github.com/auto-impl-rs/auto_impl/issues/93
 impl<T: EthereumPubSub + ?Sized> EthereumPubSub for Arc<T> {
     type SubscriptionError = T::SubscriptionError;
-    type NewHeadsStream<'a> = T::NewHeadsStream<'a> where Self: 'a;
-    type LogsStream<'a> = T::LogsStream<'a> where Self: 'a;
+    type NewHeadsStream<'a>
+        = T::NewHeadsStream<'a>
+    where
+        Self: 'a;
+    type LogsStream<'a>
+        = T::LogsStream<'a>
+    where
+        Self: 'a;
 
     fn new_heads<'a, 'async_trait>(
         &'a self,
@@ -370,8 +382,14 @@ impl<T: EthereumPubSub + ?Sized> EthereumPubSub for Arc<T> {
 
 impl<T: EthereumPubSub + ?Sized> EthereumPubSub for Box<T> {
     type SubscriptionError = T::SubscriptionError;
-    type NewHeadsStream<'a> = T::NewHeadsStream<'a> where Self: 'a;
-    type LogsStream<'a> = T::LogsStream<'a> where Self: 'a;
+    type NewHeadsStream<'a>
+        = T::NewHeadsStream<'a>
+    where
+        Self: 'a;
+    type LogsStream<'a>
+        = T::LogsStream<'a>
+    where
+        Self: 'a;
 
     fn new_heads<'a, 'async_trait>(
         &'a self,

chains/ethereum/server/src/lib.rs

@@ -113,7 +113,10 @@ impl MaybeWsEthereumClient {
 impl BlockchainClient for MaybeWsEthereumClient {
     type MetadataParams = EthereumMetadataParams;
     type Metadata = EthereumMetadata;
-    type EventStream<'a> = BlockStreamType<DefaultClient> where Self: 'a;
+    type EventStream<'a>
+        = BlockStreamType<DefaultClient>
+    where
+        Self: 'a;
     type Call = EthQuery;
     type CallResult = EthQueryResult;
 

chains/polygon/rosetta-testing-polygon/Cargo.toml

@@ -12,6 +12,7 @@ anyhow = "1.0"
 ethers = { version = "2.0", default-features = true, features = ["abigen", "rustls", "ws"] }
 ethers-solc = "2.0"
 hex-literal = "0.4"
+rosetta-chain-testing = { path = "../../rosetta-chain-testing" }
 rosetta-client.workspace = true
 rosetta-config-ethereum.workspace = true
 rosetta-core.workspace = true

chains/polygon/rosetta-testing-polygon/src/lib.rs

@@ -43,12 +43,13 @@ mod tests {
 
     use ethers_solc::{artifacts::Source, CompilerInput, EvmVersion, Solc};
     use hex_literal::hex;
+    use rosetta_chain_testing::run_test;
     use rosetta_client::Wallet;
     use rosetta_config_ethereum::{AtBlock, CallResult};
     use rosetta_core::BlockchainClient;
     use rosetta_server_ethereum::MaybeWsEthereumClient;
     use sha3::Digest;
-    use std::{collections::BTreeMap, future::Future, path::Path};
+    use std::{collections::BTreeMap, path::Path};
 
     /// Polygon rpc url
     const POLYGON_RPC_WS_URL: &str = "ws://127.0.0.1:8546";
@@ -62,38 +63,6 @@ mod tests {
         }
     }
 
-    /// Run the test in another thread while sending txs to force polygon to mine new blocks
-    /// # Panic
-    /// Panics if the future panics
-    async fn run_test<Fut: Future<Output = ()> + Send + 'static>(future: Fut) {
-        // Guarantee that only one test is incrementing blocks at a time
-        static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
-
-        // Run the test in another thread
-        let test_handler: tokio::task::JoinHandle<()> = tokio::spawn(future);
-
-        // Acquire Lock
-        let guard = LOCK.lock().await;
-
-        // Check if the test is finished after acquiring the lock
-        if test_handler.is_finished() {
-            // Release lock
-            drop(guard);
-
-            // Now is safe to panic
-            if let Err(err) = test_handler.await {
-                std::panic::resume_unwind(err.into_panic());
-            }
-            return;
-        }
-
-        // Now is safe to panic
-        if let Err(err) = test_handler.await {
-            // Resume the panic on the main task
-            std::panic::resume_unwind(err.into_panic());
-        }
-    }
-
     #[ignore = "No Polygon CI"]
     #[tokio::test]
     async fn network_status() {

chains/rosetta-chain-testing/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "rosetta-chain-testing"
+version = "0.1.0"
+edition = "2021"
+license = "MIT"
+repository = "https://github.com/analog-labs/chain-connectors"
+description = "utils for rosetta test."
+
+[dependencies]
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }

chains/rosetta-chain-testing/src/lib.rs

@@ -0,0 +1,33 @@
+use std::future::Future;
+
+/// Run the test in another thread while sending txs
+/// # Panic
+/// Panics if the future panics
+pub async fn run_test<Fut: Future<Output = ()> + Send + 'static>(future: Fut) {
+    // Guarantee that only one test is incrementing blocks at a time
+    static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+    // Run the test in another thread
+    let test_handler = tokio::spawn(future);
+
+    // Acquire Lock
+    let guard = LOCK.lock().await;
+
+    // Check if the test is finished after acquiring the lock
+    if test_handler.is_finished() {
+        // Release lock
+        drop(guard);
+
+        // Now is safe to panic
+        if let Err(err) = test_handler.await {
+            std::panic::resume_unwind(err.into_panic());
+        }
+        return;
+    }
+
+    // Now is safe to panic
+    if let Err(err) = test_handler.await {
+        // Resume the panic on the main task
+        std::panic::resume_unwind(err.into_panic());
+    }
+}

deny.toml

@@ -101,6 +101,12 @@ ignore = [
 
   # Subxt 0.37 depends deprecated libraries: https://github.com/paritytech/subxt/issues/1608
   { id = 'RUSTSEC-2024-0370', reason = "Waiting for subxt to update their deprecated dependencies" },
+
+  # RUSTSEC-2024-0377 contains multiple soundness issues:
+  { id = 'RUSTSEC-2024-0377', reason = 'Version 1.0 fixes these issues, removes the vast majority of unsafe code, and also fixes some correctness issues and on our connector using older' },
+
+  # RUSTSEC-2023-0086 contains multiple soundness issues:
+  { id = 'RUSTSEC-2023-0086', reason = 'Version 1.0 fixes these issues, removes the vast majority of unsafe code, and also fixes some correctness issues and on our connector using older' },
 ]
 
 # This section is considered when running `cargo deny check sources`.

rosetta-server/src/ws/reconnect_impl.rs

@@ -137,9 +137,18 @@ impl<T: Config> DefaultStrategy<T> {
 impl<T: Config> Reconnect for DefaultStrategy<T> {
     type Client = T::Client;
     type ClientRef = Arc<T::Client>;
-    type ReadyFuture<'a> = ReadyOrWaitFuture<T> where Self: 'a;
-    type RestartNeededFuture<'a> = ReadyOrWaitFuture<T> where Self: 'a;
-    type ReconnectFuture<'a> = ReadyOrWaitFuture<T> where Self: 'a;
+    type ReadyFuture<'a>
+        = ReadyOrWaitFuture<T>
+    where
+        Self: 'a;
+    type RestartNeededFuture<'a>
+        = ReadyOrWaitFuture<T>
+    where
+        Self: 'a;
+    type ReconnectFuture<'a>
+        = ReadyOrWaitFuture<T>
+    where
+        Self: 'a;
 
     fn ready(&self) -> Self::ReadyFuture<'_> {
         self.acquire_client()