Ban solvers based on the settlements success rate (#3263)

# Description
Another follow-up to #3257,
which implements a suggested statistic-based approach
#3257 (comment).

# Changes
Introduces an additional DB-based validator, which searches for solvers
with less than 10%(configurable) of successful settlements among 100
last auctions(configurable).

- [ ] Updated config.
- [ ] New notification details.
- [ ] Some refactoring.

## How to test
A new SQL query test.

crates/autopilot/src/arguments.rs

@@ -268,9 +268,59 @@ pub struct DbBasedSolverParticipationGuardConfig {
     #[clap(long, env, default_value = "5m", value_parser = humantime::parse_duration)]
     pub solver_blacklist_cache_ttl: Duration,
 
+    #[clap(flatten)]
+    pub non_settling_solvers_finder_config: NonSettlingSolversFinderConfig,
+
+    #[clap(flatten)]
+    pub low_settling_solvers_finder_config: LowSettlingSolversFinderConfig,
+}
+
+#[derive(Debug, clap::Parser)]
+pub struct NonSettlingSolversFinderConfig {
+    /// Enables search of non-settling solvers.
+    #[clap(
+        id = "non_settling_solvers_blacklisting_enabled",
+        long = "non-settling-solvers-blacklisting-enabled",
+        env = "NON_SETTLING_SOLVERS_BLACKLISTING_ENABLED",
+        default_value = "true"
+    )]
+    pub enabled: bool,
+
     /// The number of last auctions to check solver participation eligibility.
-    #[clap(long, env, default_value = "3")]
-    pub solver_last_auctions_participation_count: u32,
+    #[clap(
+        id = "non_settling_last_auctions_participation_count",
+        long = "non-settling-last-auctions-participation-count",
+        env = "NON_SETTLING_LAST_AUCTIONS_PARTICIPATION_COUNT",
+        default_value = "3"
+    )]
+    pub last_auctions_participation_count: u32,
+}
+
+#[derive(Debug, clap::Parser)]
+pub struct LowSettlingSolversFinderConfig {
+    /// Enables search of non-settling solvers.
+    #[clap(
+        id = "low_settling_solvers_blacklisting_enabled",
+        long = "low-settling-solvers-blacklisting-enabled",
+        env = "LOW_SETTLING_SOLVERS_BLACKLISTING_ENABLED",
+        default_value = "true"
+    )]
+    pub enabled: bool,
+
+    /// The number of last auctions to check solver participation eligibility.
+    #[clap(
+        id = "low_settling_last_auctions_participation_count",
+        long = "low-settling-last-auctions-participation-count",
+        env = "LOW_SETTLING_LAST_AUCTIONS_PARTICIPATION_COUNT",
+        default_value = "100"
+    )]
+    pub last_auctions_participation_count: u32,
+
+    /// A max failure rate for a solver to remain eligible for
+    /// participation in the competition. Otherwise, the solver will be
+    /// banned.
+    #[clap(long, env, default_value = "0.9")]
+    pub solver_max_settlement_failure_rate: f64,
 }
 
 impl std::fmt::Display for Arguments {

crates/autopilot/src/domain/competition/participation_guard/db.rs

@@ -1,44 +1,52 @@
 use {
     crate::{
+        arguments::{
+            DbBasedSolverParticipationGuardConfig,
+            LowSettlingSolversFinderConfig,
+            NonSettlingSolversFinderConfig,
+        },
         domain::{Metrics, eth},
-        infra,
+        infra::{self, solvers::dto},
     },
-    chrono::Utc,
+    chrono::{DateTime, Utc},
     ethrpc::block_stream::CurrentBlockWatcher,
     std::{
-        collections::HashMap,
+        collections::{HashMap, HashSet},
         sync::Arc,
         time::{Duration, Instant},
     },
+    tokio::join,
 };
 
 /// Checks the DB by searching for solvers that won N last consecutive auctions
-/// but never settled any of them.
+/// and either never settled any of them or their settlement success rate is
+/// lower than `min_settlement_success_rate`.
 #[derive(Clone)]
-pub(super) struct Validator(Arc<Inner>);
+pub(super) struct SolverValidator(Arc<Inner>);
 
 struct Inner {
     persistence: infra::Persistence,
     banned_solvers: dashmap::DashMap<eth::Address, Instant>,
     ttl: Duration,
-    last_auctions_count: u32,
+    non_settling_config: NonSettlingSolversFinderConfig,
+    low_settling_config: LowSettlingSolversFinderConfig,
     drivers_by_address: HashMap<eth::Address, Arc<infra::Driver>>,
 }
 
-impl Validator {
+impl SolverValidator {
     pub fn new(
         persistence: infra::Persistence,
         current_block: CurrentBlockWatcher,
         competition_updates_receiver: tokio::sync::mpsc::UnboundedReceiver<()>,
-        ttl: Duration,
-        last_auctions_count: u32,
+        db_based_validator_config: DbBasedSolverParticipationGuardConfig,
         drivers_by_address: HashMap<eth::Address, Arc<infra::Driver>>,
     ) -> Self {
         let self_ = Self(Arc::new(Inner {
             persistence,
             banned_solvers: Default::default(),
-            ttl,
-            last_auctions_count,
+            ttl: db_based_validator_config.solver_blacklist_cache_ttl,
+            non_settling_config: db_based_validator_config.non_settling_solvers_finder_config,
+            low_settling_config: db_based_validator_config.low_settling_solvers_finder_config,
             drivers_by_address,
         }));
 
@@ -59,51 +67,125 @@ impl Validator {
         tokio::spawn(async move {
             while competition_updates_receiver.recv().await.is_some() {
                 let current_block = current_block.borrow().number;
-                let non_settling_solvers = match self_
-                    .0
-                    .persistence
-                    .find_non_settling_solvers(self_.0.last_auctions_count, current_block)
-                    .await
-                {
-                    Ok(non_settling_solvers) => non_settling_solvers,
-                    Err(err) => {
-                        tracing::warn!(?err, "error while searching for non-settling solvers");
-                        continue;
-                    }
-                };
-
-                let now = Instant::now();
+
+                let (non_settling_solvers, mut low_settling_solvers) = join!(
+                    self_.find_non_settling_solvers(current_block),
+                    self_.find_low_settling_solvers(current_block)
+                );
+                // Non-settling issue has a higher priority, remove duplicates from low-settling
+                // solvers.
+                low_settling_solvers.retain(|solver| !non_settling_solvers.contains(solver));
+
+                let found_at = Instant::now();
                 let banned_until = Utc::now() + self_.0.ttl;
-                let non_settling_solver_names: Vec<&str> = non_settling_solvers
-                    .iter()
-                    .filter_map(|solver| self_.0.drivers_by_address.get(solver))
-                    .map(|driver| {
-                        Metrics::get()
-                            .non_settling_solver
-                            .with_label_values(&[&driver.name]);
-                        // Check if solver accepted this feature. This should be removed once the
-                        // CIP making this mandatory has been approved.
-                        if driver.requested_timeout_on_problems {
-                            tracing::debug!(solver = ?driver.name, "disabling solver temporarily");
-                            infra::notify_non_settling_solver(driver.clone(), banned_until);
-                            self_
-                                .0
-                                .banned_solvers
-                                .insert(driver.submission_address, now);
-                        }
-                        driver.name.as_ref()
-                    })
-                    .collect();
-
-                tracing::debug!(solvers = ?non_settling_solver_names, "found non-settling solvers");
+
+                self_.post_process(
+                    &non_settling_solvers,
+                    dto::notify::BanReason::UnsettledConsecutiveAuctions,
+                    found_at,
+                    current_block,
+                    banned_until,
+                );
+                self_.post_process(
+                    &low_settling_solvers,
+                    dto::notify::BanReason::HighSettleFailureRate,
+                    found_at,
+                    current_block,
+                    banned_until,
+                );
             }
             tracing::error!("stream of settlement updates terminated unexpectedly");
         });
     }
+
+    async fn find_non_settling_solvers(&self, current_block: u64) -> HashSet<eth::Address> {
+        if !self.0.non_settling_config.enabled {
+            return Default::default();
+        }
+
+        match self
+            .0
+            .persistence
+            .find_non_settling_solvers(
+                self.0.non_settling_config.last_auctions_participation_count,
+                current_block,
+            )
+            .await
+        {
+            Ok(solvers) => solvers.into_iter().collect(),
+            Err(err) => {
+                tracing::warn!(?err, "error while searching for non-settling solvers");
+                Default::default()
+            }
+        }
+    }
+
+    async fn find_low_settling_solvers(&self, current_block: u64) -> HashSet<eth::Address> {
+        if !self.0.low_settling_config.enabled {
+            return Default::default();
+        }
+
+        match self
+            .0
+            .persistence
+            .find_low_settling_solvers(
+                self.0.low_settling_config.last_auctions_participation_count,
+                current_block,
+                self.0
+                    .low_settling_config
+                    .solver_max_settlement_failure_rate,
+            )
+            .await
+        {
+            Ok(solvers) => solvers.into_iter().collect(),
+            Err(err) => {
+                tracing::warn!(?err, "error while searching for low-settling solvers");
+                Default::default()
+            }
+        }
+    }
+
+    /// Updates the cache and notifies the solvers.
+    fn post_process(
+        &self,
+        solvers: &HashSet<eth::Address>,
+        ban_reason: dto::notify::BanReason,
+        found_at_timestamp: Instant,
+        found_at_block: u64,
+        banned_until: DateTime<Utc>,
+    ) {
+        let non_settling_solver_names: Vec<&str> = solvers
+            .iter()
+            .filter_map(|solver| self.0.drivers_by_address.get(solver))
+            .map(|driver| {
+                Metrics::get()
+                    .banned_solver
+                    .with_label_values(&[driver.name.as_ref(), ban_reason.as_str()]);
+                // Check if solver accepted this feature. This should be removed once the
+                // CIP making this mandatory has been approved.
+                if driver.requested_timeout_on_problems {
+                    tracing::debug!(solver = ?driver.name, "disabling solver temporarily");
+                    infra::notify_banned_solver(driver.clone(), ban_reason, banned_until);
+                    self.0
+                        .banned_solvers
+                        .insert(driver.submission_address, found_at_timestamp);
+                }
+                driver.name.as_ref()
+            })
+            .collect();
+
+        let log_message = match ban_reason {
+            dto::notify::BanReason::UnsettledConsecutiveAuctions => "found non-settling solvers",
+            dto::notify::BanReason::HighSettleFailureRate => {
+                "found high-failure-settlement solvers"
+            }
+        };
+        tracing::debug!(solvers = ?non_settling_solver_names, ?found_at_block, log_message);
+    }
 }
 
 #[async_trait::async_trait]
-impl super::Validator for Validator {
+impl super::SolverValidator for SolverValidator {
     async fn is_allowed(&self, solver: &eth::Address) -> anyhow::Result<bool> {
         if let Some(entry) = self.0.banned_solvers.get(solver) {
             return Ok(entry.elapsed() >= self.0.ttl);

crates/autopilot/src/domain/competition/participation_guard/mod.rs

@@ -13,7 +13,7 @@ pub struct SolverParticipationGuard(Arc<Inner>);
 
 struct Inner {
     /// Stores the validators in order they will be called.
-    validators: Vec<Box<dyn Validator + Send + Sync>>,
+    validators: Vec<Box<dyn SolverValidator + Send + Sync>>,
 }
 
 impl SolverParticipationGuard {
@@ -24,23 +24,20 @@ impl SolverParticipationGuard {
         db_based_validator_config: DbBasedSolverParticipationGuardConfig,
         drivers: impl IntoIterator<Item = Arc<infra::Driver>>,
     ) -> Self {
-        let mut validators: Vec<Box<dyn Validator + Send + Sync>> = Vec::new();
+        let mut validators: Vec<Box<dyn SolverValidator + Send + Sync>> = Vec::new();
 
-        if db_based_validator_config.enabled {
-            let current_block = eth.current_block().clone();
-            let database_solver_participation_validator = db::Validator::new(
-                persistence,
-                current_block,
-                competition_updates_receiver,
-                db_based_validator_config.solver_blacklist_cache_ttl,
-                db_based_validator_config.solver_last_auctions_participation_count,
-                drivers
-                    .into_iter()
-                    .map(|driver| (driver.submission_address, driver.clone()))
-                    .collect(),
-            );
-            validators.push(Box::new(database_solver_participation_validator));
-        }
+        let current_block = eth.current_block().clone();
+        let database_solver_participation_validator = db::SolverValidator::new(
+            persistence,
+            current_block,
+            competition_updates_receiver,
+            db_based_validator_config,
+            drivers
+                .into_iter()
+                .map(|driver| (driver.submission_address, driver.clone()))
+                .collect(),
+        );
+        validators.push(Box::new(database_solver_participation_validator));
 
         let onchain_solver_participation_validator = onchain::Validator { eth };
         validators.push(Box::new(onchain_solver_participation_validator));
@@ -65,6 +62,6 @@ impl SolverParticipationGuard {
 }
 
 #[async_trait::async_trait]
-trait Validator: Send + Sync {
+trait SolverValidator: Send + Sync {
     async fn is_allowed(&self, solver: &eth::Address) -> anyhow::Result<bool>;
 }

crates/autopilot/src/domain/competition/participation_guard/onchain.rs

@@ -7,7 +7,7 @@ pub(super) struct Validator {
 }
 
 #[async_trait::async_trait]
-impl super::Validator for Validator {
+impl super::SolverValidator for Validator {
     async fn is_allowed(&self, solver: &eth::Address) -> anyhow::Result<bool> {
         Ok(self
             .eth

crates/autopilot/src/domain/mod.rs

@@ -18,10 +18,10 @@ pub use {
 #[derive(prometheus_metric_storage::MetricStorage)]
 #[metric(subsystem = "domain")]
 pub struct Metrics {
-    /// How many times the solver was marked as non-settling based on the
-    /// database statistics.
-    #[metric(labels("solver"))]
-    pub non_settling_solver: prometheus::IntCounterVec,
+    /// How many times the solver marked as non-settling based on the database
+    /// statistics.
+    #[metric(labels("solver", "reason"))]
+    pub banned_solver: prometheus::IntCounterVec,
 }
 
 impl Metrics {

crates/autopilot/src/infra/mod.rs

@@ -7,5 +7,5 @@ pub use {
     blockchain::Ethereum,
     order_validation::banned,
     persistence::Persistence,
-    solvers::{Driver, notify_non_settling_solver},
+    solvers::{Driver, notify_banned_solver},
 };