Colocation rc

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/uniswap/v2.rs

@@ -150,11 +150,11 @@ where
             init_code_digest: config.pool_code.into(),
         };
 
-        let pool_fetcher = PoolFetcher::new(
-            reader(web3.clone(), pair_provider),
-            web3.clone(),
-            config.missing_pool_cache_time,
-        );
+        let pool_fetcher = PoolFetcher {
+            pool_reader: reader(web3.clone(), pair_provider),
+            web3: web3.clone(),
+            non_existent_pools: Default::default(),
+        };
 
         let pool_cache = Arc::new(PoolCache::new(
             boundary::liquidity::cache_config(),

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,16 +310,18 @@ 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;
+                .map(|(trader, token, source, interactions)| {
+                    let token_contract = tokens.get(&token);
+                    let token_contract = token_contract.expect("all tokens where 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
         .into_iter()

crates/shared/src/rate_limiter.rs

@@ -174,7 +174,7 @@ impl RateLimiter {
     }
 }
 
-#[derive(Error, Debug, Clone, Default)]
+#[derive(Error, Debug, Clone, Default, PartialEq)]
 pub enum RateLimiterError {
     #[default]
     #[error("rate limited")]
@@ -221,6 +221,30 @@ impl RateLimiter {
 
         Ok(result)
     }
+
+    pub async fn execute_with_back_off<T>(
+        &self,
+        task: impl Future<Output = T>,
+        requires_back_off: impl Fn(&T) -> bool,
+    ) -> Result<T, RateLimiterError> {
+        if let Some(back_off_duration) = self.get_back_off_duration_if_limited() {
+            tokio::time::sleep(back_off_duration).await;
+        }
+
+        self.execute(task, requires_back_off).await
+    }
+
+    fn get_back_off_duration_if_limited(&self) -> Option<Duration> {
+        let strategy = self.strategy.lock().unwrap();
+        let now = Instant::now();
+
+        if strategy.drop_requests_until > now {
+            let back_off_duration = strategy.drop_requests_until - now;
+            Some(back_off_duration)
+        } else {
+            None
+        }
+    }
 }
 
 /// Shared module with common back-off checks.
@@ -236,7 +260,7 @@ pub mod back_off {
 
 #[cfg(test)]
 mod tests {
-    use {super::*, futures::FutureExt, tokio::time::sleep};
+    use {super::*, futures::FutureExt, std::ops::Add, tokio::time::sleep};
 
     #[test]
     fn current_back_off_does_not_panic() {
@@ -317,4 +341,60 @@ mod tests {
             rate_limiter.strategy().get_current_back_off()
         );
     }
+
+    #[tokio::test]
+    async fn test_execute_with_no_back_off() {
+        let timeout = Duration::from_secs(30);
+        let strategy = RateLimitingStrategy::try_new(1.0, timeout, timeout).unwrap();
+        let original_drop_until = strategy.drop_requests_until;
+        let rate_limiter = RateLimiter::from_strategy(strategy, "test_no_back_off".to_string());
+
+        let result = rate_limiter
+            .execute_with_back_off(async { 1 }, |_| false)
+            .await
+            .unwrap();
+
+        assert_eq!(result, 1);
+        {
+            let current_strategy = rate_limiter.strategy.lock().unwrap();
+            assert!(current_strategy.drop_requests_until < original_drop_until.add(timeout));
+        }
+
+        let result = rate_limiter.execute(async { 1 }, |_| false).await.unwrap();
+        assert_eq!(result, 1);
+    }
+
+    #[tokio::test]
+    async fn test_execute_with_back_off() {
+        let timeout = Duration::from_secs(3);
+        let strategy = RateLimitingStrategy::try_new(1.0, timeout, timeout).unwrap();
+        let original_drop_until = strategy.drop_requests_until;
+        let rate_limiter = RateLimiter::from_strategy(strategy, "test_back_off".to_string());
+
+        let result = rate_limiter
+            .execute_with_back_off(async { 1 }, |_| true)
+            .await
+            .unwrap();
+
+        assert_eq!(result, 1);
+        let drop_until = {
+            let current_strategy = rate_limiter.strategy.lock().unwrap();
+            let drop_until = current_strategy.drop_requests_until;
+            assert!(drop_until >= original_drop_until.add(timeout));
+            drop_until
+        };
+
+        let result = rate_limiter.execute(async { 1 }, |_| false).await;
+        assert_eq!(result, Err(RateLimiterError::RateLimited));
+        {
+            let current_strategy = rate_limiter.strategy.lock().unwrap();
+            assert_eq!(current_strategy.drop_requests_until, drop_until);
+        }
+
+        let result = rate_limiter
+            .execute_with_back_off(async { 1 }, |_| false)
+            .await
+            .unwrap();
+        assert_eq!(result, 1);
+    }
 }