Make metrics based bad order detection order specific (#4021)

# Description
Currently the bad token detection assumes that we are perfectly able to
detect "broken" orders and only orders that trade specific tokens that a
particular solver is not able to handle cause problems. However this
assumption does not work well with the increasing complexity of new
order types that can suddenly start failing for any number of reasons.
The most prominent recent example were flashloan orders where the EIP
1271 signature verified correctly but transferring the tokens into the
settlement contract failed because the user's Aave debt position was not
healthy enough.
Our current logic caused a lot of collateral damage because such orders
could cause many reasonable tokens to be flagged as unsupported although
the tokens themselves were perfectly fine and only that particular order
was problematic.

# Changes
To address this this PR change the metrics based detection mechanism to
only flag on an order by order basis instead flagging all orders trading
specific tokens. The change itself is relatively simple (collect metrics
keyed by `Uid` instead of token`) but came with a few related changes:
* the name `bad_token_detection` is now incorrect in most (but not all!)
cases so many things were renamed
* this includes a few config parameters so they must be updated in the
infra repo as well!
* caching uids has a lot more potential to bloat the cache so a cache
eviction task was introduced, this required 2 new config parameters
(max_age, gc_interval)

## How to test
- adjusted existing unit to make sure the metrics logic still works
correctly with `Uid`
- added a new unit test for the cache eviction

## Related issues

Fixes #4019

---------

Co-authored-by: Marcin Szymczak <mail@marszy.com>
Co-authored-by: ilya <ilya@cow.fi>

Author: 3 people

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

@@ -37,9 +37,9 @@ use {
 };
 
 pub mod auction;
-pub mod bad_tokens;
 pub mod order;
 mod pre_processing;
+pub mod risk_detector;
 pub mod solution;
 pub mod sorting;
 
@@ -65,7 +65,8 @@ pub struct Competition {
     pub mempools: Mempools,
     /// Cached solutions with the most recent solutions at the front.
     pub settlements: Mutex<VecDeque<Settlement>>,
-    pub bad_tokens: Arc<bad_tokens::Detector>,
+    /// bad token and orders detector
+    pub risk_detector: Arc<risk_detector::Detector>,
     fetcher: Arc<pre_processing::DataAggregator>,
     settle_queue: mpsc::Sender<SettleRequest>,
     order_sorting_strategies: Vec<Arc<dyn sorting::SortingStrategy>>,
@@ -80,7 +81,7 @@ impl Competition {
         liquidity_sources_notifier: infra::notify::liquidity_sources::Notifier,
         simulator: Simulator,
         mempools: Mempools,
-        bad_tokens: Arc<bad_tokens::Detector>,
+        risk_detector: Arc<risk_detector::Detector>,
         fetcher: Arc<DataAggregator>,
         order_sorting_strategies: Vec<Arc<dyn sorting::SortingStrategy>>,
     ) -> Arc<Self> {
@@ -95,7 +96,7 @@ impl Competition {
             mempools,
             settlements: Default::default(),
             settle_queue: settle_sender,
-            bad_tokens,
+            risk_detector,
             fetcher,
             order_sorting_strategies,
         });
@@ -233,7 +234,10 @@ impl Competition {
             .into_iter()
             .map(|solution| async move {
                 let id = solution.id().clone();
-                let token_pairs = solution.token_pairs();
+                let orders: Vec<_> = solution
+                    .user_trades()
+                    .map(|trade| trade.order().uid)
+                    .collect();
                 observe::encoding(&id);
                 let settlement = solution
                     .encode(
@@ -243,19 +247,19 @@ impl Competition {
                         self.solver.solver_native_token(),
                     )
                     .await;
-                (id, token_pairs, settlement)
+                (id, orders, settlement)
             })
             .collect::<FuturesUnordered<_>>()
-            .filter_map(|(id, token_pairs, result)| async move {
+            .filter_map(|(id, orders, result)| async move {
                 match result {
                     Ok(solution) => {
-                        self.bad_tokens.encoding_succeeded(&token_pairs);
+                        self.risk_detector.encoding_succeeded(&orders);
                         Some(solution)
                     }
                     // don't report on errors coming from solution merging
                     Err(_err) if id.solutions().len() > 1 => None,
                     Err(err) => {
-                        self.bad_tokens.encoding_failed(&token_pairs);
+                        self.risk_detector.encoding_failed(&orders);
                         observe::encoding_failed(self.solver.name(), &id, &err);
                         notify::encoding_failed(&self.solver, auction.id(), &id, &err);
                         None
@@ -741,7 +745,7 @@ impl Competition {
         if !self.solver.config().flashloans_enabled {
             auction.orders.retain(|o| o.app_data.flashloan().is_none());
         }
-        self.bad_tokens
+        self.risk_detector
             .filter_unsupported_orders_in_auction(auction)
             .await
     }

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

@@ -208,16 +208,32 @@ impl From<&solvers_dto::solution::OrderUid> for Uid {
 }
 
 impl Uid {
+    pub fn from_parts(order_hash: eth::B256, owner: eth::Address, valid_to: u32) -> Self {
+        let mut bytes = [0; UID_LEN];
+        bytes[0..32].copy_from_slice(order_hash.as_slice());
+        bytes[32..52].copy_from_slice(owner.as_slice());
+        bytes[52..56].copy_from_slice(&valid_to.to_be_bytes());
+        Self(Bytes(bytes))
+    }
+
+    /// Address that authorized the order. Sell tokens will be taken
+    /// from that address.
     pub fn owner(&self) -> eth::Address {
         self.parts().1
     }
 
+    /// Returns a UNIX timestamp after which the settlement
+    /// contract will not allow the order to be settled anymore.
+    pub fn valid_to(&self) -> u32 {
+        self.parts().2
+    }
+
     /// Splits an order UID into its parts.
     fn parts(&self) -> (eth::B256, eth::Address, u32) {
         (
             eth::B256::from_slice(&self.0.0[0..32]),
             eth::Address::from_slice(&self.0.0[32..52]),
-            u32::from_le_bytes(self.0.0[52..].try_into().unwrap()),
+            u32::from_be_bytes(self.0.0[52..].try_into().unwrap()),
         )
     }
 }