Observe multiple contracts for ETH flow

crates/autopilot/src/arguments.rs

@@ -37,10 +37,13 @@ pub struct Arguments {
     #[clap(flatten)]
     pub price_estimation: price_estimation::Arguments,
 
-    /// Address of the ethflow contract. If not specified, eth-flow orders are
+    /// Address of the ethflow contracts. If not specified, eth-flow orders are
     /// disabled.
-    #[clap(long, env)]
-    pub ethflow_contract: Option<H160>,
+    /// In general, one contract is sufficient for the service to function.
+    /// Support for multiple contract was added to support transition period for
+    /// integrators when the migration of the eth-flow contract happens.
+    #[clap(long, env, use_value_delimiter = true)]
+    pub ethflow_contracts: Vec<H160>,
 
     /// Timestamp at which we should start indexing eth-flow contract events.
     /// If there are already events in the database for a date later than this,
@@ -253,7 +256,7 @@ impl std::fmt::Display for Arguments {
             token_owner_finder,
             price_estimation,
             tracing_node_url,
-            ethflow_contract,
+            ethflow_contracts,
             ethflow_indexing_start,
             metrics_address,
             skip_event_sync,
@@ -295,7 +298,7 @@ impl std::fmt::Display for Arguments {
         write!(f, "{}", token_owner_finder)?;
         write!(f, "{}", price_estimation)?;
         display_option(f, "tracing_node_url", tracing_node_url)?;
-        writeln!(f, "ethflow_contract: {:?}", ethflow_contract)?;
+        writeln!(f, "ethflow_contracts: {:?}", ethflow_contracts)?;
         writeln!(f, "ethflow_indexing_start: {:?}", ethflow_indexing_start)?;
         writeln!(f, "metrics_address: {}", metrics_address)?;
         let _intentionally_ignored = db_url;

crates/autopilot/src/database/ethflow_events/event_retriever.rs

@@ -12,20 +12,26 @@ const ORDER_REFUND_TOPIC: H256 = H256(hex!(
 
 pub struct EthFlowRefundRetriever {
     web3: Web3,
-    address: H160,
+    addresses: Vec<H160>,
 }
 
 impl EthFlowRefundRetriever {
-    pub fn new(web3: Web3, address: H160) -> Self {
-        Self { web3, address }
+    pub fn new(web3: Web3, addresses: Vec<H160>) -> Self {
+        assert!(
+            !addresses.is_empty(),
+            "EthFlowRefundRetriever must have at least one address to listen to."
+        );
+        Self { web3, addresses }
     }
 }
 
 impl EventRetrieving for EthFlowRefundRetriever {
     type Event = contracts::cowswap_eth_flow::Event;
 
     fn get_events(&self) -> AllEventsBuilder<DynTransport, Self::Event> {
-        let mut events = AllEventsBuilder::new(self.web3.clone(), self.address, None);
+        let mut events = AllEventsBuilder::new(self.web3.clone(), H160::default(), None);
+        // We want to observe multiple addresses for events.
+        events.filter = events.filter.address(self.addresses.clone());
         // Filter out events that we don't want to listen for in the contract. `Self` is
         // designed to only pick up refunding events. Adding a filter also makes
         // the query more efficient.

crates/autopilot/src/database/onchain_order_events/event_retriever.rs

@@ -23,20 +23,26 @@ static ALL_VALID_ONCHAIN_ORDER_TOPICS: [H256; 2] =
 // onchain-order contract ABI).
 pub struct CoWSwapOnchainOrdersContract {
     web3: Web3,
-    address: H160,
+    addresses: Vec<H160>,
 }
 
 impl CoWSwapOnchainOrdersContract {
-    pub fn new(web3: Web3, address: H160) -> Self {
-        Self { web3, address }
+    pub fn new(web3: Web3, addresses: Vec<H160>) -> Self {
+        assert!(
+            !addresses.is_empty(),
+            "CoWSwapOnchainOrdersContract must have at least one address to listen to."
+        );
+        Self { web3, addresses }
     }
 }
 
 impl EventRetrieving for CoWSwapOnchainOrdersContract {
     type Event = cowswap_onchain_orders::Event;
 
     fn get_events(&self) -> AllEventsBuilder<DynTransport, Self::Event> {
-        let mut events = AllEventsBuilder::new(self.web3.clone(), self.address, None);
+        let mut events = AllEventsBuilder::new(self.web3.clone(), H160::default(), None);
+        // We want to observe multiple addresses for events.
+        events.filter = events.filter.address(self.addresses.clone());
         // Filter out events that don't belong to the ABI of `OnchainOrdersContract`.
         // This is done because there could be other unrelated events fired by
         // the contract which should be ignored. Also, it makes the request more

crates/autopilot/src/run.rs

@@ -481,7 +481,7 @@ pub async fn run(args: Arguments) {
     let mut maintenance = Maintenance::new(settlement_event_indexer, db.clone());
     maintenance.with_cow_amms(&cow_amm_registry);
 
-    if let Some(ethflow_contract) = args.ethflow_contract {
+    if !args.ethflow_contracts.is_empty() {
         let ethflow_refund_start_block = determine_ethflow_refund_indexing_start(
             &skip_event_sync_start,
             args.ethflow_indexing_start,
@@ -494,7 +494,7 @@ pub async fn run(args: Arguments) {
         let refund_event_handler = EventUpdater::new_skip_blocks_before(
             // This cares only about ethflow refund events because all the other ethflow
             // events are already indexed by the OnchainOrderParser.
-            EthFlowRefundRetriever::new(web3.clone(), ethflow_contract),
+            EthFlowRefundRetriever::new(web3.clone(), args.ethflow_contracts.clone()),
             db.clone(),
             block_retriever.clone(),
             ethflow_refund_start_block,
@@ -523,7 +523,7 @@ pub async fn run(args: Arguments) {
         let onchain_order_indexer = EventUpdater::new_skip_blocks_before(
             // The events from the ethflow contract are read with the more generic contract
             // interface called CoWSwapOnchainOrders.
-            CoWSwapOnchainOrdersContract::new(web3.clone(), ethflow_contract),
+            CoWSwapOnchainOrdersContract::new(web3.clone(), args.ethflow_contracts),
             onchain_order_event_parser,
             block_retriever,
             ethflow_start_block,

crates/e2e/src/setup/deploy.rs

@@ -26,7 +26,7 @@ pub struct Contracts {
     pub weth: WETH9,
     pub allowance: Address,
     pub domain_separator: DomainSeparator,
-    pub ethflow: CoWSwapEthFlow,
+    pub ethflows: Vec<CoWSwapEthFlow>,
     pub hooks: HooksTrampoline,
     pub cow_amm_helper: Option<CowAmmLegacyHelper>,
 }
@@ -70,7 +70,7 @@ impl Contracts {
                     .expect("Couldn't query domain separator")
                     .0,
             ),
-            ethflow: CoWSwapEthFlow::deployed(web3).await.unwrap(),
+            ethflows: vec![CoWSwapEthFlow::deployed(web3).await.unwrap()],
             hooks: HooksTrampoline::deployed(web3).await.unwrap(),
             gp_settlement,
             cow_amm_helper,
@@ -157,6 +157,7 @@ impl Contracts {
         );
 
         let ethflow = deploy!(CoWSwapEthFlow(gp_settlement.address(), weth.address()));
+        let ethflow_secondary = deploy!(CoWSwapEthFlow(gp_settlement.address(), weth.address()));
         let hooks = deploy!(HooksTrampoline(gp_settlement.address()));
 
         Self {
@@ -171,7 +172,7 @@ impl Contracts {
             weth,
             allowance,
             domain_separator,
-            ethflow,
+            ethflows: vec![ethflow, ethflow_secondary],
             hooks,
             // Current helper contract only works in forked tests
             cow_amm_helper: None,

crates/e2e/src/setup/services.rs

@@ -147,12 +147,19 @@ impl<'a> Services<'a> {
     /// deadline in case the solution would start to revert at some point)
     pub async fn start_autopilot(&self, solve_deadline: Option<Duration>, extra_args: Vec<String>) {
         let solve_deadline = solve_deadline.unwrap_or(Duration::from_secs(2));
+        let ethflow_contracts = self
+            .contracts
+            .ethflows
+            .iter()
+            .map(|c| format!("{:?}", c.address()))
+            .collect::<Vec<_>>()
+            .join(",");
 
         let args = [
             "autopilot".to_string(),
             "--max-run-loop-delay=100ms".to_string(),
             "--run-loop-native-price-timeout=500ms".to_string(),
-            format!("--ethflow-contract={:?}", self.contracts.ethflow.address()),
+            format!("--ethflow-contracts={ethflow_contracts}"),
             "--skip-event-sync=true".to_string(),
             format!("--solve-deadline={solve_deadline:?}"),
         ]