Fix NonExistentPoolMap (#2853)

# Description
We have been using
[HashSetDelay](https://docs.rs/delay_map/latest/delay_map/hashset_delay/struct.HashSetDelay.html)
incorrectly. The expiration time is only respected if the set is turned
into a stream and awaited on (it will return `Err` for expired items).

This interface is very clunky and doesn't really serve our use case.

# Changes
<!-- List of detailed changes (how the change is accomplished) -->

- [ ] Replace HashSetDelay with TtlCache which has a much more intuitive
interface

## How to test
Existing tests pass. Also, in the next PR I will add an e2e test that
actually relies on the non-existent pool cache to be reset (as liquidity
appears throughout the lifetime of the test)

crates/shared/Cargo.toml

@@ -20,7 +20,7 @@ chrono = { workspace = true, features = ["clock"] }
 clap = { workspace = true }
 contracts = { path = "../contracts" }
 database = { path = "../database" }
-delay_map = "0.3"
+ttl_cache = "0.5"
 derivative = { workspace = true }
 ethcontract = { workspace = true }
 ethrpc = { path = "../ethrpc" }

crates/shared/src/sources/uniswap_v2.rs

@@ -116,11 +116,8 @@ impl UniV2BaselineSourceParameters {
             PoolReadingStyle::Default => Box::new(pool_reader),
             PoolReadingStyle::Swapr => Box::new(SwaprPoolReader(pool_reader)),
         };
-        let fetcher = pool_fetching::PoolFetcher {
-            pool_reader,
-            web3: web3.clone(),
-            non_existent_pools: Default::default(),
-        };
+        let fetcher =
+            pool_fetching::PoolFetcher::new(pool_reader, web3.clone(), Default::default());
         Ok(UniV2BaselineSource {
             router,
             pair_provider,

crates/shared/src/sources/uniswap_v2/pool_fetching.rs

@@ -8,7 +8,6 @@ use {
     },
     anyhow::Result,
     contracts::{IUniswapLikePair, ERC20},
-    delay_map::HashSetDelay,
     ethcontract::{errors::MethodError, BlockId, H160, U256},
     futures::{
         future::{self, BoxFuture},
@@ -17,6 +16,7 @@ use {
     model::TokenPair,
     num::rational::Ratio,
     std::{collections::HashSet, sync::RwLock, time::Duration},
+    ttl_cache::TtlCache,
 };
 
 const POOL_SWAP_GAS_COST: usize = 60_000;
@@ -187,15 +187,17 @@ impl BaselineSolvable for Pool {
 pub struct PoolFetcher<Reader> {
     pub pool_reader: Reader,
     pub web3: Web3,
-    pub non_existent_pools: RwLock<HashSetDelay<TokenPair>>,
+    pub cache_time: Duration,
+    pub non_existent_pools: RwLock<TtlCache<TokenPair, ()>>,
 }
 
 impl<Reader> PoolFetcher<Reader> {
     pub fn new(reader: Reader, web3: Web3, cache_time: Duration) -> Self {
         Self {
             pool_reader: reader,
             web3,
-            non_existent_pools: RwLock::new(HashSetDelay::new(cache_time)),
+            cache_time,
+            non_existent_pools: RwLock::new(TtlCache::new(usize::MAX)),
         }
     }
 }
@@ -231,7 +233,7 @@ where
             tracing::debug!(token_pairs = ?new_missing_pairs, "stop indexing liquidity");
             let mut non_existent_pools = self.non_existent_pools.write().unwrap();
             for pair in new_missing_pairs {
-                non_existent_pools.insert(pair);
+                non_existent_pools.insert(pair, (), self.cache_time);
             }
         }
         Ok(pools)