Merge branch 'main' into shadow_metrics

Cargo.toml

@@ -10,7 +10,7 @@ cached = { version = "0.44", default-features = false }
 chrono = { version = "0.4", default-features = false }
 clap = { version = "4", features = ["derive", "env"] }
 derivative = "2"
-ethcontract = { version = "0.25.2", default-features = false, features = ["aws-kms"] }
+ethcontract = { version = "0.25.4", default-features = false, features = ["aws-kms"] }
 ethcontract-generate = { version = "0.25", default-features = false }
 ethcontract-mock = { version = "0.25", default-features = false }
 ethereum-types = "0.14"

crates/autopilot/src/database/onchain_order_events/ethflow_events.rs

@@ -143,6 +143,30 @@ fn convert_to_quote_id_and_user_valid_to(
     Ok((quote_id, user_valid_to))
 }
 
+/// The block from which to start indexing eth-flow events. Note that this
+/// function is expected to be used at the start of the services and will panic
+/// if it cannot retrieve the information it needs.
+pub async fn determine_ethflow_indexing_start(
+    skip_event_sync_start: &Option<BlockNumberHash>,
+    ethflow_indexing_start: Option<u64>,
+    web3: &Web3,
+    chain_id: u64,
+) -> BlockNumberHash {
+    if let Some(block_number_hash) = skip_event_sync_start {
+        return *block_number_hash;
+    }
+    if let Some(block_number) = ethflow_indexing_start {
+        return block_number_to_block_number_hash(web3, block_number.into())
+            .await
+            .expect("Should be able to find block at specified indexing start");
+    }
+    settlement_deployment_block_number_hash(web3, chain_id)
+        .await
+        .unwrap_or_else(|err| {
+            panic!("Should be able to find settlement deployment block. Error: {err}")
+        })
+}
+
 #[cfg(test)]
 mod test {
     use {
@@ -229,27 +253,3 @@ mod test {
         assert_eq!(result.len(), 0);
     }
 }
-
-/// The block from which to start indexing eth-flow events. Note that this
-/// function is expected to be used at the start of the services and will panic
-/// if it cannot retrieve the information it needs.
-pub async fn determine_ethflow_indexing_start(
-    skip_event_sync_start: &Option<BlockNumberHash>,
-    ethflow_indexing_start: Option<u64>,
-    web3: &Web3,
-    chain_id: u64,
-) -> BlockNumberHash {
-    if let Some(block_number_hash) = skip_event_sync_start {
-        return *block_number_hash;
-    }
-    if let Some(block_number) = ethflow_indexing_start {
-        return block_number_to_block_number_hash(web3, block_number.into())
-            .await
-            .expect("Should be able to find block at specified indexing start");
-    }
-    settlement_deployment_block_number_hash(web3, chain_id)
-        .await
-        .unwrap_or_else(|err| {
-            panic!("Should be able to find settlement deployment block. Error: {err}")
-        })
-}

crates/database/src/orders.rs

@@ -971,7 +971,7 @@ mod tests {
         let post_interactions = read_order_interactions(&mut db, &order.uid, ExecutionTime::Post)
             .await
             .unwrap();
-        assert_eq!(*post_interactions.get(0).unwrap(), post_interaction_0);
+        assert_eq!(*post_interactions.first().unwrap(), post_interaction_0);
         assert_eq!(*post_interactions.get(1).unwrap(), post_interaction_1);
 
         let post_interaction_overwrite_0 = Interaction {
@@ -992,7 +992,7 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(
-            *post_interactions.get(0).unwrap(),
+            *post_interactions.first().unwrap(),
             post_interaction_overwrite_0
         );
         assert_eq!(*post_interactions.get(1).unwrap(), post_interaction_1);
@@ -1061,7 +1061,7 @@ mod tests {
         let pre_interactions = read_order_interactions(&mut db, &order.uid, ExecutionTime::Pre)
             .await
             .unwrap();
-        assert_eq!(*pre_interactions.get(0).unwrap(), pre_interaction_0);
+        assert_eq!(*pre_interactions.first().unwrap(), pre_interaction_0);
         assert_eq!(*pre_interactions.get(1).unwrap(), pre_interaction_1);
 
         let pre_interaction_overwrite_0 = Interaction {
@@ -1082,7 +1082,7 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(
-            *pre_interactions.get(0).unwrap(),
+            *pre_interactions.first().unwrap(),
             pre_interaction_overwrite_0
         );
         assert_eq!(*pre_interactions.get(1).unwrap(), pre_interaction_1);

crates/driver/src/boundary/liquidity/mod.rs

@@ -38,7 +38,7 @@ const BLOCK_POLL_INTERVAL: Duration = Duration::from_secs(1);
 fn cache_config() -> CacheConfig {
     CacheConfig {
         number_of_blocks_to_cache: NonZeroU64::new(10).unwrap(),
-        number_of_entries_to_auto_update: NonZeroUsize::new(200).unwrap(),
+        number_of_entries_to_auto_update: NonZeroUsize::new(1000).unwrap(),
         maximum_recent_block_age: 4,
         max_retries: 5,
         delay_between_retries: Duration::from_secs(1),

crates/driver/src/boundary/liquidity/swapr.rs

@@ -51,12 +51,7 @@ pub async fn collector(
             pool_code: config.pool_code,
             missing_pool_cache_time: config.missing_pool_cache_time,
         },
-        |web3, pair_provider| {
-            SwaprPoolReader(DefaultPoolReader {
-                web3,
-                pair_provider,
-            })
-        },
+        |web3, pair_provider| SwaprPoolReader(DefaultPoolReader::new(web3, pair_provider)),
     )
     .await
 }

crates/driver/src/boundary/liquidity/uniswap/v2.rs

@@ -121,13 +121,7 @@ pub async fn collector(
     blocks: &CurrentBlockStream,
     config: &infra::liquidity::config::UniswapV2,
 ) -> Result<Box<dyn LiquidityCollecting>> {
-    collector_with_reader(eth, blocks, config, |web3, pair_provider| {
-        DefaultPoolReader {
-            web3,
-            pair_provider,
-        }
-    })
-    .await
+    collector_with_reader(eth, blocks, config, DefaultPoolReader::new).await
 }
 
 pub(in crate::boundary::liquidity) async fn collector_with_reader<R, F>(

crates/driver/src/domain/competition/auction.rs

@@ -283,6 +283,7 @@ impl AuctionProcessor {
 
     /// Fetches the tradable balance for every order owner.
     async fn fetch_balances(ethereum: &infra::Ethereum, orders: &[order::Order]) -> Balances {
+        let mut tokens: HashMap<_, _> = Default::default();
         // Collect trader/token/source/interaction tuples for fetching available
         // balances. Note that we are pessimistic here, if a trader is selling
         // the same token with the same source in two different orders using a
@@ -297,6 +298,7 @@ impl AuctionProcessor {
             .map(|((trader, token, source), mut orders)| {
                 let first = orders.next().expect("group contains at least 1 order");
                 let mut others = orders;
+                tokens.entry(token).or_insert_with(|| ethereum.erc20(token));
                 if others.all(|order| order.pre_interactions == first.pre_interactions) {
                     (trader, token, source, &first.pre_interactions[..])
                 } else {
@@ -308,15 +310,19 @@ impl AuctionProcessor {
         join_all(
             traders
                 .into_iter()
-                .map(|(trader, token, source, interactions)| async move {
-                    let balance = ethereum
-                        .erc20(token)
-                        .tradable_balance(trader.into(), source, interactions)
-                        .await;
-                    (
-                        (trader, token, source),
-                        balance.map(order::SellAmount::from).ok(),
-                    )
+                .map(|(trader, token, source, interactions)| {
+                    let token_contract = tokens.get(&token);
+                    let token_contract = token_contract.expect("all tokens were created earlier");
+                    let fetch_balance =
+                        token_contract.tradable_balance(trader.into(), source, interactions);
+
+                    async move {
+                        let balance = fetch_balance.await;
+                        (
+                            (trader, token, source),
+                            balance.map(order::SellAmount::from).ok(),
+                        )
+                    }
                 }),
         )
         .await

crates/driver/src/domain/competition/solution/mod.rs

@@ -305,11 +305,6 @@ pub enum Error {
     Boundary(#[from] boundary::Error),
     #[error("simulation error: {0:?}")]
     Simulation(#[from] simulator::Error),
-    #[error(
-        "invalid asset flow: token amounts entering the settlement do not equal token amounts \
-         exiting the settlement"
-    )]
-    AssetFlow(HashMap<eth::TokenAddress, num::BigInt>),
     #[error(transparent)]
     Execution(#[from] trade::ExecutionError),
     #[error(

crates/driver/src/domain/competition/solution/settlement.rs

@@ -1,5 +1,5 @@
 use {
-    super::{trade, Error, Solution},
+    super::{Error, Solution},
     crate::{
         boundary,
         domain::{
@@ -8,9 +8,7 @@ use {
             mempools,
         },
         infra::{blockchain::Ethereum, observe, Simulator},
-        util::conv::u256::U256Ext,
     },
-    bigdecimal::Signed,
     futures::future::try_join_all,
     std::collections::{BTreeSet, HashMap, HashSet},
 };
@@ -29,9 +27,6 @@ use {
 /// - Simulation: the settlement has been simulated without reverting, including
 ///   the case where no interactions were internalized. Additionally the solver
 ///   account is known to have sufficient Ether to execute the transaction.
-/// - Asset flow: the sum of tokens into and out of the settlement are
-///   non-negative, meaning that the solver doesn't take any tokens out of the
-///   settlement contract.
 /// - Internalization: internalized interactions only use trusted tokens.
 ///
 /// Publishing a settlement which violates these rules would result in slashing
@@ -63,44 +58,6 @@ impl Settlement {
         // For a settlement to be valid, the solution has to respect some rules which
         // would otherwise lead to slashing. Check those rules first.
 
-        // Asset flow rule: check that the sum of tokens entering the settlement is not
-        // less than the sum of tokens exiting the settlement.
-        let mut flow: HashMap<eth::TokenAddress, num::BigInt> = Default::default();
-
-        // Interaction inputs represent flow out of the contract, i.e. negative flow.
-        for input in solution
-            .interactions
-            .iter()
-            .flat_map(|interaction| interaction.inputs())
-        {
-            *flow.entry(input.token).or_default() -= eth::U256::from(input.amount).to_big_int();
-        }
-
-        // Interaction outputs represent flow into the contract, i.e. positive flow.
-        for output in solution
-            .interactions
-            .iter()
-            .flat_map(|interaction| interaction.outputs())
-        {
-            *flow.entry(output.token).or_default() += eth::U256::from(output.amount).to_big_int();
-        }
-
-        // For trades, the sold amounts are always entering the contract (positive
-        // flow), whereas the bought amounts are always exiting the contract
-        // (negative flow).
-        for trade in solution.trades.iter() {
-            let trade::Execution { sell, buy } = trade.execution(&solution)?;
-            *flow.entry(sell.token).or_default() += eth::U256::from(sell.amount).to_big_int();
-            // Within the settlement contract, the orders which buy ETH are wrapped into
-            // WETH, and hence contribute to WETH flow.
-            *flow.entry(buy.token.wrap(solution.weth)).or_default() -=
-                eth::U256::from(buy.amount).to_big_int();
-        }
-
-        if flow.values().any(|v| v.is_negative()) {
-            return Err(Error::AssetFlow(flow));
-        }
-
         // Internalization rule: check that internalized interactions only use trusted
         // tokens.
         let untrusted_tokens = solution