driver: parallelize unsupported order detection

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,46 @@ 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;
 
-        // 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()),
-                }
-            },
-            self.without_unsupported_orders(auction)
-        );
+        // Run bad token filtering and liquidity fetching in separate tasks so they can
+        // make progress independently and may execute in parallel.
+        let risk_detector = self.risk_detector.clone();
+        let flashloans_enabled = self.solver.config().flashloans_enabled;
+        let filter_auction = tokio::spawn(async move {
+            risk_detector
+                .filter_unsupported_orders_in_auction(&mut auction, flashloans_enabled)
+                .await;
+            auction
+        });
+
+        let fetch_liquidity = match self.solver.liquidity() {
+            solver::Liquidity::Fetch => {
+                let liquidity = tasks.liquidity;
+                tokio::spawn(liquidity)
+            }
+            solver::Liquidity::Skip => tokio::spawn(async { Arc::new(Vec::new()) }),
+        };
+
+        let (auction, liquidity) = tokio::join!(filter_auction, fetch_liquidity);
+        let auction = auction.map_err(|err| {
+            tracing::error!(?err, "unsupported order filtering task failed");
+            Error::PreparationError
+        })?;
+        let liquidity = liquidity.map_err(|err| {
+            tracing::error!(?err, "liquidity fetching task failed");
+            Error::PreparationError
+        })?;
 
         let elapsed = start.elapsed();
         metrics::get()
@@ -951,16 +969,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 +1085,6 @@ pub enum Error {
     NoValidOrdersFound,
     #[error("could not parse the request")]
     MalformedRequest,
+    #[error("Preparation task failed")]
+    PreparationError,
 }