refactor: parallelize unsupported-order detection and defer filtering

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

@@ -1,10 +1,6 @@
 use {
     self::solution::settlement,
-    super::{
-        Mempools,
-        mempools::SubmissionMode,
-        time::{self, Remaining},
-    },
+    super::{Mempools, mempools::SubmissionMode},
     crate::{
         domain::{
             competition::{solution::Settlement, sorting::SortingStrategy},
@@ -47,7 +43,7 @@ pub mod solution;
 pub mod sorting;
 
 use {
-    crate::infra::notify::liquidity_sources::LiquiditySourceNotifying,
+    crate::{domain::time::Remaining, infra::notify::liquidity_sources::LiquiditySourceNotifying},
     eth_domain_types::BlockNo,
 };
 pub use {auction::Auction, order::Order, pre_processing::DataAggregator, solution::Solution};
@@ -220,16 +216,16 @@ pub struct Competition {
     pub solver: Solver,
     pub eth: Ethereum,
     pub liquidity: infra::liquidity::Fetcher,
-    pub liquidity_sources_notifier: infra::notify::liquidity_sources::Notifier,
+    pub liquidity_sources_notifier: notify::liquidity_sources::Notifier,
     pub simulator: Simulator,
     pub mempools: Mempools,
     /// Cached solutions with the most recent solutions at the front.
     pub settlements: Mutex<VecDeque<Settlement>>,
     /// bad token and orders detector
     pub risk_detector: Arc<risk_detector::Detector>,
-    fetcher: Arc<pre_processing::DataAggregator>,
+    fetcher: Arc<DataAggregator>,
     settle_queue: mpsc::Sender<SettleRequest>,
-    order_sorting_strategies: Vec<Arc<dyn sorting::SortingStrategy>>,
+    order_sorting_strategies: Vec<Arc<dyn SortingStrategy>>,
     submitter_pool: SubmitterPool,
 }
 
@@ -239,12 +235,12 @@ impl Competition {
         solver: Solver,
         eth: Ethereum,
         liquidity: infra::liquidity::Fetcher,
-        liquidity_sources_notifier: infra::notify::liquidity_sources::Notifier,
+        liquidity_sources_notifier: notify::liquidity_sources::Notifier,
         simulator: Simulator,
         mempools: Mempools,
         risk_detector: Arc<risk_detector::Detector>,
         fetcher: Arc<DataAggregator>,
-        order_sorting_strategies: Vec<Arc<dyn sorting::SortingStrategy>>,
+        order_sorting_strategies: Vec<Arc<dyn SortingStrategy>>,
     ) -> Arc<Self> {
         let submission_accounts = solver.submission_accounts().to_vec();
         if !submission_accounts.is_empty() {
@@ -296,50 +292,62 @@ impl Competition {
                 tracing::error!(?err, "pre-processing auction failed");
                 Error::MalformedRequest
             })?;
+
         let mut auction = Arc::unwrap_or_clone(tasks.auction.await);
 
         let solver_address = self.solver.address();
         let order_sorting_strategies = self.order_sorting_strategies.clone();
 
-        // Add the CoW AMM orders to the auction
+        // Add CoW AMM orders to the auction
         let cow_amm_orders = tasks.cow_amm_orders.await;
         auction.orders.extend(cow_amm_orders.iter().cloned());
 
+        // Start unsupported-order detection immediately
+        let orders_for_unsupported = auction.orders.clone();
+        let unsupported_orders_future =
+            async move { self.unsupported_order_uids(&orders_for_unsupported).await };
+
         let sort_orders_future = Self::run_blocking_with_timer("sort_orders", move || {
-            // Use spawn_blocking() because a lot of CPU bound computations are happening
+            // Use spawn_blocking() because a lot of CPU bound computations are happening,
             // and we don't want to block the runtime for too long.
             Self::sort_orders(auction, solver_address, order_sorting_strategies)
         });
 
-        // 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);
+        // We can sort the orders, determine UIDS to filter and fetch auction data in
+        // parallel.
+        let (auction, balances, app_data, unsupported_uids) = tokio::join!(
+            sort_orders_future,
+            tasks.balances,
+            tasks.app_data,
+            unsupported_orders_future,
+        );
 
-        let 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.
+        // 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.
+        let mut auction = Self::run_blocking_with_timer("update_orders", move || {
             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)
-        );
+        // Apply unsupported filtering after update_orders.
+        if !unsupported_uids.is_empty() {
+            auction
+                .orders
+                .retain(|order| !unsupported_uids.contains(&order.uid));
+        }
+
+        let liquidity = match self.solver.liquidity() {
+            solver::Liquidity::Fetch => tasks.liquidity.await,
+            solver::Liquidity::Skip => Arc::new(Vec::new()),
+        };
 
         let elapsed = start.elapsed();
         metrics::get()
             .auction_preprocessing
             .with_label_values(&["total"])
             .observe(elapsed.as_secs_f64());
         drop(timer);
+
         tracing::debug!(?elapsed, "auction task execution time");
 
         if auction.orders.is_empty() {
@@ -570,8 +578,8 @@ impl Competition {
         Ok(score)
     }
 
-    // Oders already need to be sorted from most relevant to least relevant so that
-    // we allocate balances for the most relevants first.
+    // Orders already need to be sorted from most relevant to the least relevant so
+    // that we allocate balances for the most relevant first.
     fn sort_orders(
         mut auction: Auction,
         solver: eth::Address,
@@ -603,7 +611,7 @@ impl Competition {
 
         // The auction that we receive from the `autopilot` assumes that there
         // is sufficient balance to completely cover all the orders. **This is
-        // not the case** (as the protocol should not chose which limit orders
+        // not the case** (as the protocol should not choose which limit orders
         // get filled for some given sell token balance). This loop goes through
         // the priority sorted orders and allocates the available user balance
         // to each order, and potentially scaling the order's `available` amount
@@ -613,7 +621,7 @@ impl Competition {
             if cow_amms.contains(&order.uid) {
                 // cow amm orders already get constructed fully initialized
                 // so we don't have to handle them here anymore.
-                // Without this short circuiting logic they would get filtered
+                // Without this short-circuit logic they would get filtered
                 // out later because we don't bother fetching their balances
                 // for performance reasons.
                 return true;
@@ -938,13 +946,20 @@ impl Competition {
     }
 
     #[instrument(skip_all)]
-    async fn without_unsupported_orders(&self, mut auction: Auction) -> Auction {
+    async fn unsupported_order_uids(&self, orders: &[Order]) -> HashSet<order::Uid> {
+        let mut removed = HashSet::new();
+
         if !self.solver.config().flashloans_enabled {
-            auction.orders.retain(|o| o.app_data.flashloan().is_none());
+            removed.extend(
+                orders
+                    .iter()
+                    .filter_map(|o| o.app_data.flashloan().is_some().then_some(o.uid)),
+            );
         }
-        self.risk_detector
-            .filter_unsupported_orders_in_auction(auction)
-            .await
+
+        removed.extend(self.risk_detector.unsupported_order_uids(orders).await);
+
+        removed
     }
 }
 
@@ -1059,7 +1074,7 @@ pub enum Error {
     )]
     SolutionNotAvailable,
     #[error("{0:?}")]
-    DeadlineExceeded(#[from] time::DeadlineExceeded),
+    DeadlineExceeded(#[from] DeadlineExceeded),
     #[error("solver error: {0:?}")]
     Solver(#[from] solver::Error),
     #[error("failed to submit the solution")]

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

@@ -6,20 +6,25 @@
 //! was simply built with a buggy compiler which makes it incompatible
 //! with the settlement contract (see <https://github.com/cowprotocol/services/pull/781>).
 //!
-//! Additionally there are some heuristics to detect when an
+//! Additionally, there are some heuristics to detect when an
 //! order itself is somehow broken or causes issues and slipped through
 //! other detection mechanisms. One big error case is orders adjusting
-//! debt postions in lending protocols. While pre-checks might correctly
+//! debt positions in lending protocols. While pre-checks might correctly
 //! detect that the EIP 1271 signature is valid the transfer of the token
 //! would fail because the user's debt position is not collateralized enough.
-//! In other words the bad order detection is a last fail safe in case
+//! In other words the bad order detection is a last fail-safe in case
 //! we were not able to predict issues with orders and pre-emptively
 //! filter them out of the auction.
+
 use {
-    crate::domain::competition::{Auction, order::Uid},
+    crate::domain::competition::{Order, order::Uid},
     eth_domain_types as eth,
     futures::{StreamExt, stream::FuturesUnordered},
-    std::{collections::HashMap, fmt, time::Instant},
+    std::{
+        collections::{HashMap, HashSet},
+        fmt,
+        time::Instant,
+    },
 };
 
 pub mod bad_orders;
@@ -80,63 +85,73 @@ impl Detector {
         self
     }
 
-    /// Removes all unsupported orders from the auction.
-    pub async fn filter_unsupported_orders_in_auction(&self, mut auction: Auction) -> Auction {
-        let now = Instant::now();
+    /// Filters unsupported orders out of the auction.
+    pub async fn filter_unsupported_orders_in_auction(
+        &self,
+        mut auction: crate::domain::competition::Auction,
+    ) -> crate::domain::competition::Auction {
+        let removed_uids = self.unsupported_order_uids(&auction.orders).await;
+        if !removed_uids.is_empty() {
+            auction
+                .orders
+                .retain(|order| !removed_uids.contains(&order.uid));
+        }
+        auction
+    }
 
-        // reuse the original allocation
-        let supported_orders = std::mem::take(&mut auction.orders);
+    /// Returns the UIDs of orders this solver cannot support.
+    pub async fn unsupported_order_uids(&self, orders: &[Order]) -> HashSet<Uid> {
+        let now = Instant::now();
         let mut token_quality_checks = FuturesUnordered::new();
-        let mut removed_uids = Vec::new();
-
-        let mut supported_orders: Vec<_> = supported_orders
-            .into_iter()
-            .filter(|order| {
-                self.metrics
-                    .as_ref()
-                    .map(|metrics| metrics.get_quality(&order.uid, now))
-                    .is_none_or(|q| q != Quality::Unsupported)
-            })
-            .filter_map(|order| {
-                let sell = self.get_token_quality(order.sell.token, now);
-                let buy = self.get_token_quality(order.buy.token, now);
-                match (sell, buy) {
-                    // both tokens supported => keep order
-                    (Quality::Supported, Quality::Supported) => Some(order),
-                    // at least 1 token unsupported => drop order
-                    (Quality::Unsupported, _) | (_, Quality::Unsupported) => {
-                        removed_uids.push(order.uid);
-                        None
-                    }
-                    // sell token quality is unknown => keep order if token is supported
-                    (Quality::Unknown, _) => {
-                        let Some(detector) = &self.simulation_detector else {
-                            // we can't determine quality => assume order is good
-                            return Some(order);
-                        };
-                        let check_tokens_fut = async move {
-                            let quality = detector.determine_sell_token_quality(&order, now).await;
-                            (order, quality)
-                        };
-                        token_quality_checks.push(check_tokens_fut);
-                        None
-                    }
-                    // buy token quality is unknown => keep order (because we can't
-                    // determine quality and assume it's good)
-                    (_, Quality::Unknown) => Some(order),
+        let mut removed_uids = HashSet::new();
+
+        for order in orders {
+            if self
+                .metrics
+                .as_ref()
+                .map(|metrics| metrics.get_quality(&order.uid, now))
+                .is_some_and(|q| q == Quality::Unsupported)
+            {
+                removed_uids.insert(order.uid);
+                continue;
+            }
+
+            let sell = self.get_token_quality(order.sell.token, now);
+            let buy = self.get_token_quality(order.buy.token, now);
+
+            match (sell, buy) {
+                // sell token quality is unknown => keep order if token is supported
+                (Quality::Supported, Quality::Supported) => {}
+                // at least 1 token unsupported => drop order
+                (Quality::Unsupported, _) | (_, Quality::Unsupported) => {
+                    removed_uids.insert(order.uid);
                 }
-            })
-            .collect();
-
-        while let Some((order, quality)) = token_quality_checks.next().await {
-            if quality == Quality::Supported {
-                supported_orders.push(order);
-            } else {
-                removed_uids.push(order.uid);
+                // sell token quality is unknown => keep order if token is supported
+                (Quality::Unknown, _) => {
+                    // we can't determine quality => assume order is good
+                    let Some(detector) = &self.simulation_detector else {
+                        continue;
+                    };
+
+                    let order = order.clone();
+                    let check_tokens_fut = async move {
+                        let quality = detector.determine_sell_token_quality(&order, now).await;
+                        (order.uid, quality)
+                    };
+                    token_quality_checks.push(check_tokens_fut);
+                }
+                // buy token quality is unknown => keep order (because we can't determine quality
+                // and assume it's good)
+                (_, Quality::Unknown) => {}
+            }
+        }
+
+        while let Some((uid, quality)) = token_quality_checks.next().await {
+            if quality != Quality::Supported {
+                removed_uids.insert(uid);
             }
         }
 
-        auction.orders = supported_orders;
         if !removed_uids.is_empty() {
             tracing::debug!(orders = ?removed_uids, "ignored orders with unsupported tokens");
         }
@@ -145,7 +160,7 @@ impl Detector {
             detector.evict_outdated_entries();
         }
 
-        auction
+        removed_uids
     }
 
     /// Updates the tokens quality metric for successful operation.

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

@@ -1,11 +1,12 @@
+use std::sync::Arc;
 use {
     crate::{
         domain::competition::{auction::Tokens, order},
         util,
     },
     chrono::Duration,
     eth_domain_types as eth,
-    std::{fmt::Debug, sync::Arc},
+    std::fmt::Debug,
 };
 
 #[derive(PartialEq, Eq, PartialOrd, Ord)]
@@ -60,6 +61,7 @@ impl SortingStrategy for ExternalPrice {
 /// which is significantly faster than the
 /// [num::BigRational] we used before.
 pub struct OrdFloat(f64);
+
 impl PartialOrd for OrdFloat {
     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
         Some(self.cmp(other))