Move current block stream into ethrpc (#1820)

The overall idea is to move block streams and event streams into the `ethrpc` crate. This PR takes care of the `current_block` stream. Note that the files `shared::ethrpc` and `shared::current_block` still exist but they only contain the CLI logic. In a follow up refactoring it would make sense to move that last remaining logic into `shared::arguments`. ### Test Plan Only moved code around so the e2e tests should be enough.
cowprotocol · Aug 29, 2023 · 64ad56f · 64ad56f
1 parent d467474
commit 64ad56f
Show file tree

Hide file tree

Showing 43 changed files with 470 additions and 463 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/autopilot/Cargo.toml b/crates/autopilot/Cargo.toml
@@ -25,6 +25,7 @@ contracts = { path = "../contracts" }
 database = { path = "../database" }
 derivative = { workspace = true }
 ethcontract = { workspace = true }
+ethrpc = { path = "../ethrpc" }
 futures = { workspace = true }
 gas-estimation = { workspace = true }
 observe = { path = "../observe" }

diff --git a/crates/autopilot/src/database/ethflow_events/event_storing.rs b/crates/autopilot/src/database/ethflow_events/event_storing.rs
@@ -4,7 +4,8 @@ use {
     crate::database::{events::bytes_to_order_uid, Postgres},
     anyhow::Result,
     database::ethflow_orders::Refund,
-    shared::{current_block::RangeInclusive, event_handling::EventStoring},
+    ethrpc::current_block::RangeInclusive,
+    shared::event_handling::EventStoring,
 };
 
 fn get_refunds(events: Vec<ethcontract::Event<EthFlowEvent>>) -> Result<Vec<Refund>> {

diff --git a/crates/autopilot/src/database/events.rs b/crates/autopilot/src/database/events.rs
@@ -16,8 +16,9 @@ use {
         OrderUid,
     },
     ethcontract::{Event as EthContractEvent, EventMetadata},
+    ethrpc::current_block::RangeInclusive,
     number_conversions::u256_to_big_decimal,
-    shared::{current_block::RangeInclusive, event_handling::EventStoring},
+    shared::event_handling::EventStoring,
     std::convert::TryInto,
 };
 

diff --git a/crates/autopilot/src/database/onchain_order_events.rs b/crates/autopilot/src/database/onchain_order_events.rs
@@ -21,6 +21,10 @@ use {
         PgTransaction,
     },
     ethcontract::{Event as EthContractEvent, H160},
+    ethrpc::{
+        current_block::{timestamp_of_block_in_seconds, RangeInclusive},
+        Web3,
+    },
     futures::{stream, StreamExt},
     itertools::multiunzip,
     model::{
@@ -31,14 +35,12 @@ use {
     },
     number_conversions::u256_to_big_decimal,
     shared::{
-        current_block::{timestamp_of_block_in_seconds, RangeInclusive},
         db_order_conversions::{
             buy_token_destination_into,
             order_kind_into,
             sell_token_source_into,
             signing_scheme_into,
         },
-        ethrpc::Web3,
         event_handling::EventStoring,
         order_quoting::{OrderQuoting, Quote, QuoteSearchParameters},
         order_validation::{

diff --git a/crates/autopilot/src/database/onchain_order_events/ethflow_events.rs b/crates/autopilot/src/database/onchain_order_events/ethflow_events.rs
@@ -15,12 +15,12 @@ use {
         PgTransaction,
     },
     ethcontract::Event as EthContractEvent,
-    hex_literal::hex,
-    shared::{
-        contracts::settlement_deployment_block_number_hash,
+    ethrpc::{
         current_block::{block_number_to_block_number_hash, BlockNumberHash},
-        ethrpc::Web3,
+        Web3,
     },
+    hex_literal::hex,
+    shared::contracts::settlement_deployment_block_number_hash,
     sqlx::types::BigDecimal,
     std::{collections::HashMap, convert::TryInto},
 };

diff --git a/crates/autopilot/src/event_updater.rs b/crates/autopilot/src/event_updater.rs
@@ -1,8 +1,8 @@
 use {
     anyhow::Result,
     contracts::gpv2_settlement,
+    ethrpc::current_block::{BlockNumberHash, BlockRetrieving},
     shared::{
-        current_block::{BlockNumberHash, BlockRetrieving},
         event_handling::{EventHandler, EventRetrieving, EventStoring},
         impl_event_retrieving,
         maintenance::Maintaining,

diff --git a/crates/autopilot/src/lib.rs b/crates/autopilot/src/lib.rs
@@ -15,6 +15,7 @@ use {
     },
     contracts::{BalancerV2Vault, IUniswapV3Factory, WETH9},
     ethcontract::{errors::DeployError, BlockNumber},
+    ethrpc::current_block::block_number_to_block_number_hash,
     futures::StreamExt,
     model::DomainSeparator,
     shared::{
@@ -27,7 +28,6 @@ use {
             trace_call::TraceCallDetector,
         },
         baseline_solver::BaseTokens,
-        current_block::block_number_to_block_number_hash,
         fee_subsidy::{config::FeeSubsidyConfiguration, FeeSubsidizing},
         gas_price::InstrumentedGasEstimator,
         http_client::HttpClientFactory,

diff --git a/crates/autopilot/src/on_settlement_event_updater.rs b/crates/autopilot/src/on_settlement_event_updater.rs
@@ -41,13 +41,9 @@ use {
     anyhow::{anyhow, Context, Result},
     contracts::GPv2Settlement,
     database::byte_array::ByteArray,
+    ethrpc::{current_block::CurrentBlockStream, Web3},
     primitive_types::{H160, H256},
-    shared::{
-        current_block::CurrentBlockStream,
-        ethrpc::Web3,
-        event_handling::MAX_REORG_BLOCK_COUNT,
-        external_prices::ExternalPrices,
-    },
+    shared::{event_handling::MAX_REORG_BLOCK_COUNT, external_prices::ExternalPrices},
     sqlx::PgConnection,
     std::time::Duration,
     web3::types::{Transaction, TransactionId},
@@ -297,7 +293,7 @@ mod tests {
         database::clear_DANGER(&db.0).await.unwrap();
         let transport = shared::ethrpc::create_env_test_transport();
         let web3 = Web3::new(transport);
-        let current_block = shared::current_block::current_block_stream(
+        let current_block = ethrpc::current_block::current_block_stream(
             Arc::new(web3.clone()),
             Duration::from_secs(1),
         )

diff --git a/crates/autopilot/src/run_loop.rs b/crates/autopilot/src/run_loop.rs
@@ -14,6 +14,7 @@ use {
     anyhow::{anyhow, ensure, Context, Result},
     chrono::Utc,
     database::order_events::OrderEventLabel,
+    ethrpc::{current_block::CurrentBlockStream, Web3},
     itertools::Itertools,
     model::{
         auction::{Auction, AuctionId},
@@ -29,12 +30,7 @@ use {
     },
     primitive_types::{H160, H256},
     rand::seq::SliceRandom,
-    shared::{
-        current_block::CurrentBlockStream,
-        ethrpc::Web3,
-        event_handling::MAX_REORG_BLOCK_COUNT,
-        token_list::AutoUpdatingTokenList,
-    },
+    shared::{event_handling::MAX_REORG_BLOCK_COUNT, token_list::AutoUpdatingTokenList},
     std::{
         collections::{BTreeMap, HashSet},
         sync::Arc,

diff --git a/crates/autopilot/src/solvable_orders.rs b/crates/autopilot/src/solvable_orders.rs
@@ -4,6 +4,7 @@ use {
     bigdecimal::BigDecimal,
     chrono::Utc,
     database::order_events::OrderEventLabel,
+    ethrpc::current_block::CurrentBlockStream,
     itertools::Itertools,
     model::{
         auction::Auction,
@@ -18,7 +19,6 @@ use {
     shared::{
         account_balances::{BalanceFetching, Query},
         bad_token::BadTokenDetecting,
-        current_block::CurrentBlockStream,
         price_estimation::native_price_cache::CachingNativePriceEstimator,
         remaining_amounts,
         signature_validator::{SignatureCheck, SignatureValidating},

diff --git a/crates/driver/src/boundary/liquidity/balancer/v2/mod.rs b/crates/driver/src/boundary/liquidity/balancer/v2/mod.rs
@@ -17,8 +17,8 @@ use {
         BalancerV2WeightedPoolFactoryV3,
         GPv2Settlement,
     },
+    ethrpc::current_block::{BlockRetrieving, CurrentBlockStream},
     shared::{
-        current_block::{BlockRetrieving, CurrentBlockStream},
         http_solver::model::TokenAmount,
         sources::balancer_v2::{
             pool_fetching::BalancerContracts,

diff --git a/crates/driver/src/boundary/liquidity/mod.rs b/crates/driver/src/boundary/liquidity/mod.rs
@@ -5,12 +5,13 @@ use {
         infra::{self, blockchain::Ethereum},
     },
     anyhow::Result,
+    ethrpc::current_block::CurrentBlockStream,
     futures::future,
     itertools::Itertools,
     model::TokenPair,
     shared::{
         baseline_solver::BaseTokens,
-        current_block::{self, CurrentBlockStream},
+        current_block,
         http_client::HttpClientFactory,
         recent_block_cache::{self, CacheConfig},
     },

diff --git a/crates/driver/src/boundary/liquidity/swapr.rs b/crates/driver/src/boundary/liquidity/swapr.rs
@@ -4,9 +4,10 @@ use {
         domain::liquidity::{self, swapr},
         infra::{self, blockchain::Ethereum},
     },
-    shared::{
-        current_block::CurrentBlockStream,
-        sources::{swapr::reader::SwaprPoolReader, uniswap_v2::pool_fetching::DefaultPoolReader},
+    ethrpc::current_block::CurrentBlockStream,
+    shared::sources::{
+        swapr::reader::SwaprPoolReader,
+        uniswap_v2::pool_fetching::DefaultPoolReader,
     },
     solver::{liquidity::ConstantProductOrder, liquidity_collector::LiquidityCollecting},
 };

diff --git a/crates/driver/src/boundary/liquidity/uniswap/v2.rs b/crates/driver/src/boundary/liquidity/uniswap/v2.rs
@@ -9,10 +9,12 @@ use {
     },
     async_trait::async_trait,
     contracts::{GPv2Settlement, IUniswapLikeRouter},
-    ethrpc::Web3,
+    ethrpc::{
+        current_block::{self, CurrentBlockStream},
+        Web3,
+    },
     futures::StreamExt,
     shared::{
-        current_block::{self, CurrentBlockStream},
         http_solver::model::TokenAmount,
         maintenance::Maintaining,
         sources::uniswap_v2::{

diff --git a/crates/driver/src/boundary/liquidity/uniswap/v3.rs b/crates/driver/src/boundary/liquidity/uniswap/v3.rs
@@ -12,9 +12,9 @@ use {
     },
     anyhow::Context,
     contracts::{GPv2Settlement, UniswapV3SwapRouter},
+    ethrpc::current_block::BlockRetrieving,
     itertools::Itertools,
     shared::{
-        current_block::BlockRetrieving,
         http_solver::model::TokenAmount,
         interaction::Interaction,
         sources::uniswap_v3::pool_fetching::UniswapV3PoolFetcher,

diff --git a/crates/e2e/Cargo.toml b/crates/e2e/Cargo.toml
@@ -15,6 +15,7 @@ contracts = { path = "../contracts" }
 database = { path = "../database" }
 driver = { path = "../driver" }
 ethcontract = { workspace = true }
+ethrpc = { path = "../ethrpc" }
 hex = { workspace = true }
 hex-literal = { workspace = true }
 model = { path = "../model" }

diff --git a/crates/e2e/tests/e2e/eth_flow.rs b/crates/e2e/tests/e2e/eth_flow.rs
@@ -5,6 +5,7 @@ use {
     contracts::{CoWSwapEthFlow, ERC20Mintable, WETH9},
     e2e::{nodes::local_node::TestNodeApi, setup::*, tx, tx_value},
     ethcontract::{transaction::TransactionResult, Account, Bytes, H160, H256, U256},
+    ethrpc::{current_block::timestamp_of_current_block_in_seconds, Web3},
     hex_literal::hex,
     model::{
         order::{
@@ -36,11 +37,7 @@ use {
         refund_service::{INVALIDATED_OWNER, NO_OWNER},
     },
     reqwest::Client,
-    shared::{
-        current_block::timestamp_of_current_block_in_seconds,
-        ethrpc::Web3,
-        signature_validator::check_erc1271_result,
-    },
+    shared::signature_validator::check_erc1271_result,
 };
 
 const DAI_PER_ETH: u32 = 1_000;

diff --git a/crates/e2e/tests/e2e/refunder.rs b/crates/e2e/tests/e2e/refunder.rs
@@ -3,9 +3,9 @@ use {
     chrono::{TimeZone, Utc},
     e2e::{nodes::local_node::TestNodeApi, setup::*},
     ethcontract::{H160, U256},
+    ethrpc::{current_block::timestamp_of_current_block_in_seconds, Web3},
     model::quote::{OrderQuoteRequest, OrderQuoteSide, QuoteSigningScheme, Validity},
     refunder::refund_service::RefundService,
-    shared::{current_block::timestamp_of_current_block_in_seconds, ethrpc::Web3},
     sqlx::PgPool,
 };
 

diff --git a/crates/ethrpc/Cargo.toml b/crates/ethrpc/Cargo.toml
@@ -11,6 +11,7 @@ path = "src/lib.rs"
 doctest = false
 
 [dependencies]
+anyhow = { workspace = true }
 async-trait = { workspace= true }
 ethereum-types = { workspace = true }
 futures = { workspace = true }
@@ -19,13 +20,15 @@ hex-literal = { workspace = true }
 lazy_static = { workspace = true }
 mockall = { workspace = true }
 observe = { path = "../observe" }
+primitive-types = { workspace = true }
 prometheus = { workspace = true }
 prometheus-metric-storage = { git = "https://github.com/cowprotocol/prometheus-metric-storage", tag = "v0.4.0" }
 reqwest = { workspace = true }
 scopeguard = "1"
 serde = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = [] }
+tokio-stream = { version = "0.1", features = ["sync"] }
 web3 = { workspace = true }
 contracts = { path = "../contracts" }
 ethcontract = { workspace = true }