Remove CoW AMM indexer from driver (#3680)

# Description
Currently, both autopilot and driver use the same CoW AMM indexer
functionality in parallel to serve different purposes: autopilot needs
to fill out surplus capturing jit order owners in the auction and driver
uses it to fetch tradable tokens and then prepare CoW AMM template
orders. Since autopilot already sends CoW AMM addresses within each
auction, the only missing part is the CoW AMM helper SC address. It can
be received by fetching the CoW AMM factory address and then using a
config mapping to find the helper address. This should also reduce the
RPC traffic.

# Changes

- [ ] Driver now contains only the mapping between AMM factory and
helper SC.
- [ ] When it receives a CoW AMM in the auction, it tries to fetch the
AMM factory address by calling the `FACTORY` function. Since we can now
completely deprecate legacy CoW AMMs, this logic is simplified. I also
used a simple abi to create rust bindings for this interface. The
fetched data is stored in the memory cache.
- [ ] Using the factory address, it gets the helper SC instance based on
the config.
- [ ] Simplify the e2e tests config.
- [ ] Adjusted a forked e2e test to start using the new factory address
with some caveats(the test with its comments should be
self-descriptive).

## How to test
Adjusted forked e2e test.

Author: squadgazzz

crates/contracts/artifacts/CowAmmFactoryGetter.json

@@ -0,0 +1,17 @@
+{
+  "abi": [
+    {
+      "inputs": [],
+      "name": "FACTORY",
+      "outputs": [
+        {
+          "internalType": "address",
+          "name": "",
+          "type": "address"
+        }
+      ],
+      "stateMutability": "view",
+      "type": "function"
+    }
+  ]
+}

crates/contracts/src/alloy.rs

@@ -556,6 +556,10 @@ crate::bindings!(
     }
 );
 
+pub mod cow_amm {
+    crate::bindings!(CowAmmFactoryGetter);
+}
+
 pub mod support {
     // Support contracts used for trade and token simulations.
     crate::bindings!(AnyoneAuthenticator);

crates/cow-amm/src/amm.rs

@@ -20,10 +20,7 @@ pub struct Amm {
 }
 
 impl Amm {
-    pub(crate) async fn new(
-        address: Address,
-        helper: &CowAmmLegacyHelper,
-    ) -> Result<Self, MethodError> {
+    pub async fn new(address: Address, helper: &CowAmmLegacyHelper) -> Result<Self, MethodError> {
         let tradeable_tokens = helper.tokens(address).call().await?;
 
         Ok(Self {

crates/driver/example.toml

@@ -44,8 +44,6 @@ flashloan-router = "0x0000000000000000000000000000000000000000"
 factory = "0x86f3df416979136cb4fdea2c0886301b911c163b"
 # address of contract to help interfacing with the created CoW AMMs
 helper = "0x86f3df416979136cb4fdea2c0886301b911c163b"
-# at which block the driver should start indexing the factory (1 block before deployment)
-index-start = 20188649
 
 [liquidity]
 base-tokens = [

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

@@ -3,6 +3,7 @@ use {
     crate::{
         domain::{
             competition::order::{SellTokenBalance, app_data::AppData},
+            cow_amm,
             eth,
             liquidity,
         },
@@ -58,6 +59,7 @@ pub struct Utilities {
     liquidity_fetcher: infra::liquidity::Fetcher,
     tokens: tokens::Fetcher,
     balance_fetcher: Arc<dyn BalanceFetching>,
+    cow_amm_cache: cow_amm::Cache,
 }
 
 impl std::fmt::Debug for Utilities {
@@ -130,6 +132,14 @@ impl DataAggregator {
             eth.balance_overrider(),
         );
 
+        let cow_amm_helper_by_factory = eth
+            .contracts()
+            .cow_amm_helper_by_factory()
+            .iter()
+            .map(|(factory, helper)| (factory.0.into(), helper.0.into()))
+            .collect();
+        let cow_amm_cache = cow_amm::Cache::new(eth.web3().clone(), cow_amm_helper_by_factory);
+
         Self {
             utilities: Arc::new(Utilities {
                 eth,
@@ -138,6 +148,7 @@ impl DataAggregator {
                 liquidity_fetcher,
                 tokens,
                 balance_fetcher,
+                cow_amm_cache,
             }),
             control: Mutex::new(ControlBlock {
                 solve_request: Default::default(),
@@ -380,17 +391,19 @@ impl Utilities {
         let _timer = metrics::get().processing_stage_timer("cow_amm_orders");
         let _timer2 =
             observe::metrics::metrics().on_auction_overhead_start("driver", "cow_amm_orders");
-        let cow_amms = self.eth.contracts().cow_amm_registry().amms().await;
+
+        let cow_amms = self
+            .cow_amm_cache
+            .get_or_create_amms(&auction.surplus_capturing_jit_order_owners)
+            .await;
+
         let domain_separator = self.eth.contracts().settlement_domain_separator();
         let domain_separator = model::DomainSeparator(domain_separator.0);
         let validator = self.signature_validator.as_ref();
+
         let results: Vec<_> = futures::future::join_all(
             cow_amms
                 .into_iter()
-                // Only generate orders for cow amms the auction told us about.
-                // Otherwise the solver would expect the order to get surplus but
-                // the autopilot would actually not count it.
-                .filter(|amm| auction.surplus_capturing_jit_order_owners.contains(&eth::Address(*amm.address())))
                 // Only generate orders where the auction provided the required
                 // reference prices. Otherwise there will be an error during the
                 // surplus calculation which will also result in 0 surplus for

crates/driver/src/domain/cow_amm.rs

@@ -0,0 +1,128 @@
+use {
+    crate::domain::eth,
+    contracts::CowAmmLegacyHelper,
+    cow_amm::Amm,
+    ethrpc::alloy::conversions::{IntoAlloy, IntoLegacy},
+    itertools::{
+        Either::{Left, Right},
+        Itertools,
+    },
+    std::{
+        collections::{HashMap, HashSet},
+        sync::Arc,
+    },
+    tokio::sync::RwLock,
+};
+
+/// Cache for CoW AMM data to avoid using the registry dependency.
+/// Maps AMM address to the corresponding Amm instance.
+pub struct Cache {
+    inner: RwLock<HashMap<eth::Address, Arc<Amm>>>,
+    web3: ethrpc::Web3,
+    helper_by_factory: HashMap<eth::Address, CowAmmLegacyHelper>,
+}
+
+impl Cache {
+    pub fn new(web3: ethrpc::Web3, factory_mapping: HashMap<eth::Address, eth::Address>) -> Self {
+        let helper_by_factory = factory_mapping
+            .into_iter()
+            .map(|(factory, helper)| (factory, CowAmmLegacyHelper::at(&web3, helper.0)))
+            .collect();
+        Self {
+            inner: RwLock::new(HashMap::new()),
+            web3,
+            helper_by_factory,
+        }
+    }
+
+    /// Gets or creates AMM instances for the given surplus capturing JIT order
+    /// owners.
+    pub async fn get_or_create_amms(
+        &self,
+        surplus_capturing_jit_order_owners: &HashSet<eth::Address>,
+    ) -> Vec<Arc<Amm>> {
+        let (mut cached_amms, missing_amms): (Vec<Arc<Amm>>, Vec<eth::Address>) = {
+            let cache = self.inner.read().await;
+            surplus_capturing_jit_order_owners
+                .iter()
+                .partition_map(|&address| match cache.get(&address) {
+                    Some(amm) => Left(amm.clone()),
+                    None => Right(address),
+                })
+        };
+
+        if missing_amms.is_empty() {
+            return cached_amms;
+        }
+
+        let fetch_futures = missing_amms.into_iter().map(|amm_address| async move {
+            let factory_address = self
+                .fetch_amm_factory_address(amm_address)
+                .await
+                .inspect_err(|err| {
+                    tracing::warn!(
+                        ?err,
+                        amm_address = ?amm_address.0,
+                        "failed to fetch CoW AMM factory address"
+                    );
+                })
+                .ok()?;
+
+            let Some(helper) = self.helper_by_factory.get(&factory_address) else {
+                tracing::warn!(
+                    factory_address = ?factory_address.0,
+                    amm_address = ?amm_address.0,
+                    "no helper contract configured for CoW AMM factory"
+                );
+                return None;
+            };
+
+            match Amm::new(amm_address.0, helper).await {
+                Ok(amm) => Some((amm_address, Arc::new(amm))),
+                Err(err) => {
+                    let helper_address = helper.address().0;
+                    tracing::warn!(
+                        ?err,
+                        amm_address = ?amm_address.0,
+                        ?helper_address,
+                        "failed to create CoW AMM instance"
+                    );
+                    None
+                }
+            }
+        });
+
+        let fetched_results = futures::future::join_all(fetch_futures).await;
+
+        // Update cache with newly fetched AMMs
+        let newly_created_amms = {
+            let mut cache = self.inner.write().await;
+            fetched_results
+                .into_iter()
+                .flatten()
+                .map(|(amm_address, amm)| {
+                    cache.insert(amm_address, amm.clone());
+                    amm
+                })
+                .collect::<Vec<_>>()
+        };
+
+        // Combine cached and newly created AMMs
+        cached_amms.extend(newly_created_amms);
+        cached_amms
+    }
+
+    /// Fetches the factory address for the given AMM by calling the
+    /// `FACTORY` function.
+    async fn fetch_amm_factory_address(
+        &self,
+        amm_address: eth::Address,
+    ) -> anyhow::Result<eth::Address> {
+        let factory_getter =
+            contracts::alloy::cow_amm::CowAmmFactoryGetter::CowAmmFactoryGetter::new(
+                amm_address.0.into_alloy(),
+                &self.web3.alloy,
+            );
+        Ok(factory_getter.FACTORY().call().await?.into_legacy().into())
+    }
+}

crates/driver/src/domain/mod.rs

@@ -1,4 +1,5 @@
 pub mod competition;
+pub mod cow_amm;
 pub mod eth;
 pub mod liquidity;
 pub mod mempools;

crates/driver/src/infra/blockchain/contracts.rs

@@ -1,12 +1,12 @@
 use {
-    crate::{boundary, domain::eth, infra::blockchain::Ethereum},
+    crate::{domain::eth, infra::blockchain::Ethereum},
     chain::Chain,
     contracts::alloy::{BalancerV2Vault, FlashLoanRouter, support::Balances},
     ethrpc::{
         Web3,
         alloy::conversions::{IntoAlloy, IntoLegacy},
-        block_stream::CurrentBlockWatcher,
     },
+    std::collections::HashMap,
     thiserror::Error,
 };
 
@@ -20,14 +20,16 @@ pub struct Contracts {
 
     /// The domain separator for settlement contract used for signing orders.
     settlement_domain_separator: eth::DomainSeparator,
-    cow_amm_registry: cow_amm::Registry,
 
     /// Single router that supports multiple flashloans in the
     /// same settlement.
     // TODO: make this non-optional when contracts are deployed
     // everywhere
     flashloan_router: Option<FlashLoanRouter::Instance>,
     balance_helper: Balances::Instance,
+    /// Mapping from CoW AMM factory address to the corresponding CoW AMM
+    /// helper.
+    cow_amm_helper_by_factory: HashMap<eth::ContractAddress, eth::ContractAddress>,
 }
 
 #[derive(Debug, Default, Clone)]
@@ -36,7 +38,7 @@ pub struct Addresses {
     pub signatures: Option<eth::ContractAddress>,
     pub weth: Option<eth::ContractAddress>,
     pub balances: Option<eth::ContractAddress>,
-    pub cow_amms: Vec<CowAmmConfig>,
+    pub cow_amm_helper_by_factory: HashMap<eth::ContractAddress, eth::ContractAddress>,
     pub flashloan_router: Option<eth::ContractAddress>,
 }
 
@@ -45,8 +47,6 @@ impl Contracts {
         web3: &Web3,
         chain: Chain,
         addresses: Addresses,
-        block_stream: CurrentBlockWatcher,
-        archive_node: Option<super::RpcArgs>,
     ) -> Result<Self, Error> {
         let address_for = |contract: &ethcontract::Contract,
                            address: Option<eth::ContractAddress>| {
@@ -99,21 +99,6 @@ impl Contracts {
                 .0,
         );
 
-        let archive_node_web3 = archive_node.as_ref().map_or(web3.clone(), |args| {
-            boundary::buffered_web3_client(
-                &args.url,
-                args.max_batch_size,
-                args.max_concurrent_requests,
-            )
-        });
-        let mut cow_amm_registry = cow_amm::Registry::new(archive_node_web3);
-        for config in addresses.cow_amms {
-            cow_amm_registry
-                .add_listener(config.index_start, config.factory, config.helper)
-                .await;
-        }
-        cow_amm_registry.spawn_maintenance_task(block_stream);
-
         // TODO: use `address_for()` once contracts are deployed
         let flashloan_router = addresses
             .flashloan_router
@@ -133,9 +118,9 @@ impl Contracts {
             signatures,
             weth,
             settlement_domain_separator,
-            cow_amm_registry,
             flashloan_router,
             balance_helper,
+            cow_amm_helper_by_factory: addresses.cow_amm_helper_by_factory,
         })
     }
 
@@ -167,27 +152,19 @@ impl Contracts {
         &self.settlement_domain_separator
     }
 
-    pub fn cow_amm_registry(&self) -> &cow_amm::Registry {
-        &self.cow_amm_registry
-    }
-
     pub fn flashloan_router(&self) -> Option<&FlashLoanRouter::Instance> {
         self.flashloan_router.as_ref()
     }
 
     pub fn balance_helper(&self) -> &Balances::Instance {
         &self.balance_helper
     }
-}
 
-#[derive(Debug, Clone)]
-pub struct CowAmmConfig {
-    /// Which contract to index for CoW AMM deployment events.
-    pub factory: eth::H160,
-    /// Which helper contract to use for interfacing with the indexed CoW AMMs.
-    pub helper: eth::H160,
-    /// At which block indexing should start on the factory.
-    pub index_start: u64,
+    pub fn cow_amm_helper_by_factory(
+        &self,
+    ) -> &HashMap<eth::ContractAddress, eth::ContractAddress> {
+        &self.cow_amm_helper_by_factory
+    }
 }
 
 /// Returns the address of a contract for the specified network, or `None` if

crates/driver/src/infra/blockchain/mod.rs

@@ -96,7 +96,6 @@ impl Ethereum {
         rpc: Rpc,
         addresses: contracts::Addresses,
         gas: Arc<GasPriceEstimator>,
-        archive_node_url: Option<&Url>,
         tx_gas_limit: U256,
     ) -> Self {
         let Rpc { web3, chain, args } = rpc;
@@ -108,19 +107,9 @@ impl Ethereum {
         .await
         .expect("couldn't initialize current block stream");
 
-        let contracts = Contracts::new(
-            &web3,
-            chain,
-            addresses,
-            current_block_stream.clone(),
-            archive_node_url.map(|url| RpcArgs {
-                url: url.clone(),
-                max_batch_size: args.max_batch_size,
-                max_concurrent_requests: args.max_concurrent_requests,
-            }),
-        )
-        .await
-        .expect("could not initialize important smart contracts");
+        let contracts = Contracts::new(&web3, chain, addresses)
+            .await
+            .expect("could not initialize important smart contracts");
         let balance_overrider = Arc::new(BalanceOverrides::new(web3.clone()));
         let balance_simulator = BalanceSimulator::new(
             contracts.settlement().clone(),