Remove CoW AMM indexer from driver

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

@@ -554,6 +554,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,
         },
@@ -62,6 +63,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 {
@@ -134,6 +136,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,
@@ -142,6 +152,7 @@ impl DataAggregator {
                 liquidity_fetcher,
                 tokens,
                 balance_fetcher,
+                cow_amm_cache,
             }),
             control: Mutex::new(ControlBlock {
                 solve_request: Default::default(),
@@ -377,17 +388,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::{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>| {
@@ -97,21 +97,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
@@ -131,9 +116,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,
         })
     }
 
@@ -165,27 +150,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(),