driver: parallelize unsupported order detection

crates/driver/Cargo.toml

@@ -92,6 +92,7 @@ app-data = { workspace = true, features = ["test_helpers"] }
 contracts = { workspace = true }
 ethrpc = { workspace = true, features = ["test-util"] }
 maplit = { workspace = true }
+mockall = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["process", "test-util"] }
 

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

@@ -33,7 +33,7 @@ use {
         time::Instant,
     },
     tokio::{sync::mpsc, task},
-    tracing::{Instrument, instrument},
+    tracing::Instrument,
 };
 
 pub mod auction;
@@ -339,28 +339,51 @@ impl Competition {
             Self::sort_orders(auction, solver_address, order_sorting_strategies)
         });
 
-        // We can sort the orders and fetch auction data in parallel
+        // We can sort the orders and fetch auction data in parallel.
         let (auction, balances, app_data) =
             tokio::join!(sort_orders_future, tasks.balances, tasks.app_data);
 
-        let auction = Self::run_blocking_with_timer("update_orders", move || {
+        let mut auction = Self::run_blocking_with_timer("update_orders", move || {
             // Same as before with sort_orders, we use spawn_blocking() because a lot of CPU
             // bound computations are happening and we want to avoid blocking
             // the runtime.
             Self::update_orders(auction, balances, app_data, cow_amm_orders)
         })
         .await;
 
+        let risk_detector = self.risk_detector.clone();
+        let flashloans_enabled = self.solver.config().flashloans_enabled;
+        let liquidity_mode = self.solver.liquidity();
+
         // We can run bad token filtering and liquidity fetching in parallel
-        let (liquidity, auction) = tokio::join!(
-            async {
-                match self.solver.liquidity() {
-                    solver::Liquidity::Fetch => tasks.liquidity.await,
-                    solver::Liquidity::Skip => Arc::new(Vec::new()),
+        let (auction, liquidity) = tokio::join!(
+            tokio::spawn(
+                async move {
+                    risk_detector
+                        .without_unsupported_orders(&mut auction.orders, flashloans_enabled)
+                        .await;
+                    auction
                 }
-            },
-            self.without_unsupported_orders(auction)
+                .in_current_span(),
+            ),
+            tokio::spawn(
+                async move {
+                    match liquidity_mode {
+                        solver::Liquidity::Fetch => tasks.liquidity.await,
+                        solver::Liquidity::Skip => Arc::new(Vec::new()),
+                    }
+                }
+                .in_current_span(),
+            ),
         );
+        let auction = auction.map_err(|err| {
+            tracing::error!(?err, "order filtering task failed");
+            Error::InternalError(err.to_string())
+        })?;
+        let liquidity = liquidity.map_err(|err| {
+            tracing::error!(?err, "liquidity fetch task failed");
+            Error::InternalError(err.to_string())
+        })?;
 
         let elapsed = start.elapsed();
         metrics::get()
@@ -951,16 +974,6 @@ impl Competition {
         }
         Ok(())
     }
-
-    #[instrument(skip_all)]
-    async fn without_unsupported_orders(&self, mut auction: Auction) -> Auction {
-        if !self.solver.config().flashloans_enabled {
-            auction.orders.retain(|o| o.app_data.flashloan().is_none());
-        }
-        self.risk_detector
-            .filter_unsupported_orders_in_auction(auction)
-            .await
-    }
 }
 
 const MAX_SOLUTIONS_TO_MERGE: usize = 10;
@@ -1077,4 +1090,6 @@ pub enum Error {
     NoValidOrdersFound,
     #[error("could not parse the request")]
     MalformedRequest,
+    #[error("internal error: {0}")]
+    InternalError(String),
 }

crates/driver/src/domain/competition/risk_detector/bad_tokens/simulation.rs

@@ -5,10 +5,12 @@ use {
         infra::{self, observe::metrics},
     },
     bad_tokens::{TokenQuality, trace_call::TraceCallDetectorRaw},
+    eth_domain_types as eth,
     futures::FutureExt,
     model::interaction::InteractionData,
     request_sharing::BoxRequestSharing,
     std::{
+        ops::Deref,
         sync::Arc,
         time::{Duration, Instant},
     },
@@ -21,6 +23,29 @@ use {
 #[derive(Clone)]
 pub struct Detector(Arc<Inner>);
 
+#[cfg_attr(test, mockall::automock)]
+#[async_trait::async_trait]
+pub trait DetectorApi: Send + Sync {
+    async fn determine_sell_token_quality(&self, order: &Order, now: Instant) -> Quality;
+    fn get_quality(&self, token: &eth::TokenAddress, now: Instant) -> Quality;
+    fn evict_outdated_entries(&self);
+}
+
+#[async_trait::async_trait]
+impl DetectorApi for Detector {
+    async fn determine_sell_token_quality(&self, order: &Order, now: Instant) -> Quality {
+        self.determine_sell_token_quality(order, now).await
+    }
+
+    fn get_quality(&self, token: &eth::TokenAddress, now: Instant) -> Quality {
+        Deref::deref(self).get_quality(token, now)
+    }
+
+    fn evict_outdated_entries(&self) {
+        Deref::deref(self).evict_outdated_entries()
+    }
+}
+
 struct Inner {
     cache: Cache,
     detector: TraceCallDetectorRaw,
@@ -107,7 +132,7 @@ impl Detector {
     }
 }
 
-impl std::ops::Deref for Detector {
+impl Deref for Detector {
     type Target = Cache;
 
     fn deref(&self) -> &Self::Target {