Metrics-based bad token detector

Cargo.toml

@@ -10,6 +10,7 @@ bigdecimal = "0.3"
 cached = { version = "0.49.3", default-features = false }
 chrono = { version = "0.4.38", default-features = false }
 clap = { version = "4.5.6", features = ["derive", "env"] }
+dashmap = "6.1.0"
 derivative = "2.2.0"
 derive_more = { version = "1.0.0", features = ["full"] }
 ethcontract = { version = "0.25.7", default-features = false, features = ["aws-kms"] }

crates/autopilot/src/run.rs

@@ -274,16 +274,16 @@ pub async fn run(args: Arguments) {
 
     let trace_call_detector = args.tracing_node_url.as_ref().map(|tracing_node_url| {
         CachingDetector::new(
-            Box::new(TraceCallDetector {
-                web3: shared::ethrpc::web3(
+            Box::new(TraceCallDetector::new(
+                shared::ethrpc::web3(
                     &args.shared.ethrpc,
                     &http_factory,
                     tracing_node_url,
                     "trace",
                 ),
+                eth.contracts().settlement().address(),
                 finder,
-                settlement_contract: eth.contracts().settlement().address(),
-            }),
+            )),
             args.shared.token_quality_cache_expiry,
             args.shared.token_quality_cache_prefetch_time,
         )

crates/driver/Cargo.toml

@@ -24,6 +24,7 @@ axum = { workspace = true }
 bigdecimal = { workspace = true }
 chrono = { workspace = true, features = ["clock"], default-features = false }
 cow-amm = { path = "../cow-amm" }
+dashmap = { workspace = true }
 derive_more = { workspace = true }
 ethabi = "18.0"
 ethereum-types = { workspace = true }

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

@@ -30,7 +30,7 @@ pub struct Auction {
     /// See the [`Self::id`] method.
     id: Option<Id>,
     /// See the [`Self::orders`] method.
-    orders: Vec<competition::Order>,
+    pub(crate) orders: Vec<competition::Order>,
     /// The tokens that are used in the orders of this auction.
     tokens: Tokens,
     gas_price: eth::GasPrice,

crates/driver/src/domain/competition/bad_tokens/cache.rs

@@ -0,0 +1,87 @@
+use {
+    crate::domain::{competition::bad_tokens::Quality, eth},
+    dashmap::{DashMap, Entry},
+    std::{
+        sync::Arc,
+        time::{Duration, Instant},
+    },
+};
+
+/// Cache keeping track of whether or not a token is considered supported or
+/// not. Internally reference counted for cheap clones and easy sharing.
+/// Stores a map instead of a set to not recompute the quality of good tokens
+/// over and over.
+/// Evicts cached value after a configurable period of time.
+#[derive(Clone)]
+pub struct Cache(Arc<Inner>);
+
+struct Inner {
+    cache: DashMap<eth::TokenAddress, CacheEntry>,
+    /// entries older than this get ignored and evicted
+    max_age: Duration,
+}
+
+struct CacheEntry {
+    /// when the decision on the token quality was made
+    timestamp: Instant,
+    /// whether the token is supported or not
+    quality: Quality,
+}
+
+impl Cache {
+    /// Creates a new instance which evicts cached values after a period of
+    /// time.
+    pub fn new(max_age: Duration) -> Self {
+        Self(Arc::new(Inner {
+            max_age,
+            cache: DashMap::default(),
+        }))
+    }
+
+    /// Updates whether or not a token should be considered supported.
+    pub fn update_quality(&self, token: eth::TokenAddress, quality: Quality, now: Instant) {
+        match self.0.cache.entry(token) {
+            Entry::Occupied(mut o) => {
+                let value = o.get_mut();
+                if now.duration_since(value.timestamp) > self.0.max_age
+                    || quality == Quality::Unsupported
+                {
+                    // Only update the value if the cached value is outdated by now or
+                    // if the new value is "Unsupported". This means on conflicting updates
+                    // we err on the conservative side and assume a token is unsupported.
+                    value.quality = quality;
+                }
+                value.timestamp = now;
+            }
+            Entry::Vacant(v) => {
+                v.insert(CacheEntry {
+                    quality,
+                    timestamp: now,
+                });
+            }
+        }
+    }
+
+    pub fn evict_outdated_entries(&self) {
+        let now = Instant::now();
+        self.0
+            .cache
+            .retain(|_, value| now.duration_since(value.timestamp) < self.0.max_age);
+    }
+
+    /// Returns the quality of the token. If the cached value is older than the
+    /// `max_age` it gets ignored and the token evicted.
+    pub fn get_quality(&self, token: eth::TokenAddress, now: Instant) -> Option<Quality> {
+        let Entry::Occupied(entry) = self.0.cache.entry(token) else {
+            return None;
+        };
+
+        let value = entry.get();
+        if now.duration_since(value.timestamp) > self.0.max_age {
+            entry.remove();
+            return None;
+        }
+
+        Some(value.quality)
+    }
+}

crates/driver/src/domain/competition/bad_tokens/metrics.rs

@@ -0,0 +1,52 @@
+use {
+    super::Quality,
+    crate::domain::eth,
+    dashmap::DashMap,
+    std::{collections::HashSet, sync::Arc},
+};
+
+/// Monitors tokens to determine whether they are considered "unsupported" based
+/// on the number of consecutive participation in failing settlement encoding.
+/// Tokens that consistently participate in failures beyond a predefined
+/// threshold are marked as unsupported. Once token participates in a successful
+/// settlement encoding, it is removed from the cache.
+#[derive(Default)]
+pub struct Detector(Arc<Inner>);
+
+#[derive(Default)]
+struct Inner {
+    counter: DashMap<eth::TokenAddress, u8>,
+}
+
+impl Detector {
+    /// Defines the threshold for the number of consecutive unsupported
+    /// quality detections before a token is considered unsupported.
+    const UNSUPPORTED_THRESHOLD: u8 = 5;
+
+    pub fn get_quality(&self, token: eth::TokenAddress) -> Option<Quality> {
+        self.0
+            .counter
+            .get(&token)
+            .filter(|counter| **counter >= Self::UNSUPPORTED_THRESHOLD)
+            .map(|_| Quality::Unsupported)
+    }
+
+    /// Increments the counter of failures for each token.
+    pub fn update_failing_tokens(&self, tokens: HashSet<eth::TokenAddress>) {
+        for token in tokens {
+            self.0
+                .counter
+                .entry(token)
+                .and_modify(|counter| *counter = Self::UNSUPPORTED_THRESHOLD.min(*counter + 1))
+                .or_insert(1);
+        }
+    }
+
+    /// Removes tokens from the cache since they all participated in a
+    /// successful settlement encoding.
+    pub fn update_successful_tokens(&self, tokens: HashSet<eth::TokenAddress>) {
+        for token in tokens {
+            self.0.counter.remove(&token);
+        }
+    }
+}

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

@@ -0,0 +1,141 @@
+use {
+    crate::domain::{competition::Auction, eth},
+    futures::StreamExt,
+    std::{
+        collections::{HashMap, HashSet},
+        fmt,
+        time::Instant,
+    },
+};
+
+pub mod cache;
+pub mod metrics;
+pub mod simulation;
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum Quality {
+    /// Solver is likely to produce working solutions when computing
+    /// routes for this token.
+    Supported,
+    /// Solver will likely produce failing solutions when computing
+    /// routes for this token. This can have many reasons:
+    /// * fees on transfer
+    /// * token enforces max transfer amount
+    /// * trader is deny listed
+    /// * bugs in the solidity compiler make it incompatible with the settlement
+    ///   contract - see <https://github.com/cowprotocol/services/pull/781>
+    /// * probably tons of other reasons
+    Unsupported,
+}
+
+#[derive(Default)]
+pub struct Detector {
+    /// manually configured list of supported and unsupported tokens. Only
+    /// tokens that get detected incorrectly by the automatic detectors get
+    /// listed here and therefore have a higher precedence.
+    hardcoded: HashMap<eth::TokenAddress, Quality>,
+    simulation_detector: Option<simulation::Detector>,
+    metrics: Option<metrics::Detector>,
+}
+
+impl Detector {
+    /// Hardcodes tokens as (un)supported based on the provided config. This has
+    /// the highest priority when looking up a token's quality.
+    pub fn new(config: HashMap<eth::TokenAddress, Quality>) -> Self {
+        Self {
+            hardcoded: config,
+            ..Default::default()
+        }
+    }
+
+    /// Enables detection of unsupported tokens via simulation based detection
+    /// methods.
+    pub fn with_simulation_detector(&mut self, detector: simulation::Detector) -> &mut Self {
+        self.simulation_detector = Some(detector);
+        self
+    }
+
+    /// Enables detection of unsupported tokens based on heuristics.
+    pub fn with_heuristic_detector(&mut self) -> &mut Self {
+        self.metrics = Some(metrics::Detector::default());
+        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();
+
+        let filtered_orders = futures::stream::iter(auction.orders.into_iter())
+            .filter_map(move |order| async move {
+                let sell = self.get_token_quality(order.sell.token, now);
+                let buy = self.get_token_quality(order.sell.token, now);
+                match (sell, buy) {
+                    // both tokens supported => keep order
+                    (Some(Quality::Supported), Some(Quality::Supported)) => Some(order),
+                    // at least 1 token unsupported => drop order
+                    (Some(Quality::Unsupported), _) | (_, Some(Quality::Unsupported)) => None,
+                    // sell token quality is unknown => keep order if token is supported
+                    (None, _) => {
+                        let Some(detector) = &self.simulation_detector else {
+                            // we can't determine quality => assume order is good
+                            return Some(order);
+                        };
+                        let quality = detector.determine_sell_token_quality(&order, now).await;
+                        (quality == Some(Quality::Supported)).then_some(order)
+                    }
+                    // buy token quality is unknown => keep order (because we can't
+                    // determine quality and assume it's good)
+                    (_, None) => Some(order),
+                }
+            })
+            .collect::<Vec<_>>()
+            .await;
+        auction.orders = filtered_orders;
+
+        if let Some(detector) = &self.simulation_detector {
+            detector.evict_outdated_entries();
+        }
+
+        auction
+    }
+
+    /// Updates the tokens quality metric for successful operation.
+    pub fn encoding_succeeded(&self, tokens: HashSet<eth::TokenAddress>) {
+        if let Some(metrics) = &self.metrics {
+            metrics.update_successful_tokens(tokens);
+        }
+    }
+
+    /// Updates the tokens quality metric for failures.
+    pub fn encoding_failed(&self, tokens: HashSet<eth::TokenAddress>) {
+        if let Some(metrics) = &self.metrics {
+            metrics.update_failing_tokens(tokens);
+        }
+    }
+
+    fn get_token_quality(&self, token: eth::TokenAddress, now: Instant) -> Option<Quality> {
+        if let Some(quality) = self.hardcoded.get(&token) {
+            return Some(*quality);
+        }
+
+        if let Some(detector) = &self.simulation_detector {
+            if let Some(quality) = detector.get_quality(token, now) {
+                return Some(quality);
+            }
+        }
+
+        if let Some(metrics) = &self.metrics {
+            return metrics.get_quality(token);
+        }
+
+        None
+    }
+}
+
+impl fmt::Debug for Detector {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Detector")
+            .field("hardcoded", &self.hardcoded)
+            .finish()
+    }
+}