Handle fatal errors when indexing cow amms (#2934)

# Description
Currently the `autopilot` maintenance loop can get stuck when the
configured cow amm helper contract does not support a found contract
with this error:
```
2024-09-03T06:13:17.420Z  WARN autopilot::maintenance: failed to run background task successfully err=method 'tokens(address):(address[])' failure: contract call reverted with message: None

Caused by:
    contract call reverted with message: None
```

So far we have been retrying this error over and over although the call
to detect which tokens the pool is trading will never work if an amm is
misconfigured or generally not supported by the helper contract.

# Changes
Differentiate between retryable and fatal errors while indexing the amms
and retry or skip this pool respectively.

# Testplan
Set up sepolia configuration locally and checked that we don't get stuck
on the unsupported pools:
```
2024-09-03T06:24:55.880Z  INFO cow_amm::cache: helper contract does not support amm cow_amm=0xe4abfda4e8c02fcafc34981dafaeb426aa4186e6 err=MethodError { signature: "tokens(address):(address[])", inner: Revert(None) }
2024-09-03T06:24:55.921Z  INFO cow_amm::cache: indexed new cow amm cow_amm=0xac140f325afd20a733e12580aeb22ff9bf46982f
2024-09-03T06:24:55.956Z  INFO cow_amm::cache: indexed new cow amm cow_amm=0xa54442606548bf1b627662a465a40b31b7e8e711
```

crates/cow-amm/src/amm.rs

@@ -2,7 +2,7 @@ use {
     anyhow::Result,
     app_data::AppDataHash,
     contracts::CowAmmLegacyHelper,
-    ethcontract::{Address, Bytes, U256},
+    ethcontract::{errors::MethodError, Address, Bytes, U256},
     model::{
         interaction::InteractionData,
         order::{BuyTokenDestination, OrderData, OrderKind, SellTokenSource},
@@ -18,7 +18,10 @@ pub struct Amm {
 }
 
 impl Amm {
-    pub(crate) async fn new(address: Address, helper: &CowAmmLegacyHelper) -> Result<Self> {
+    pub(crate) async fn new(
+        address: Address,
+        helper: &CowAmmLegacyHelper,
+    ) -> Result<Self, MethodError> {
         let tradeable_tokens = helper.tokens(address).call().await?;
 
         Ok(Self {

crates/cow-amm/src/cache.rs

@@ -1,7 +1,7 @@
 use {
     crate::Amm,
     contracts::{cow_amm_legacy_helper::Event as CowAmmEvent, CowAmmLegacyHelper},
-    ethcontract::Address,
+    ethcontract::{errors::ExecutionError, Address},
     ethrpc::block_stream::RangeInclusive,
     shared::event_handling::EventStoring,
     std::{collections::BTreeMap, sync::Arc},
@@ -76,22 +76,36 @@ impl EventStoring<CowAmmEvent> for Storage {
         &mut self,
         events: Vec<ethcontract::Event<CowAmmEvent>>,
     ) -> anyhow::Result<()> {
-        let cache = &mut *self.0.cache.write().await;
-
+        let mut processed_events = Vec::with_capacity(events.len());
         for event in events {
-            let meta = event
-                .meta
-                .as_ref()
-                .ok_or_else(|| anyhow::anyhow!("Event missing meta"))?;
+            let Some(meta) = event.meta else {
+                tracing::warn!(?event, "event does not contain required meta data");
+                continue;
+            };
+
             let CowAmmEvent::CowammpoolCreated(cow_amm) = event.data;
             let cow_amm = cow_amm.amm;
-
-            cache
-                .entry(meta.block_number)
-                .or_default()
-                .push(Arc::new(Amm::new(cow_amm, &self.0.helper).await?));
-            tracing::info!(?cow_amm, "indexed new cow amm");
+            match Amm::new(cow_amm, &self.0.helper).await {
+                Ok(amm) => processed_events.push((meta.block_number, Arc::new(amm))),
+                Err(err) if matches!(&err.inner, ExecutionError::Web3(_)) => {
+                    // Abort completely to later try the entire block range again.
+                    // That keeps the cache in a consistent state and avoids indexing
+                    // the same event multiple times which would result in duplicate amms.
+                    tracing::debug!(?cow_amm, ?err, "retryable error");
+                    return Err(err.into());
+                }
+                Err(err) => {
+                    tracing::info!(?cow_amm, ?err, "helper contract does not support amm");
+                    continue;
+                }
+            };
         }
+        let cache = &mut *self.0.cache.write().await;
+        for (block, amm) in processed_events {
+            tracing::info!(cow_amm = ?amm.address(), "indexed new cow amm");
+            cache.entry(block).or_default().push(amm);
+        }
+
         Ok(())
     }
 

crates/shared/src/arguments.rs

@@ -99,7 +99,7 @@ pub struct OrderQuotingArguments {
 
 logging_args_with_default_filter!(
     LoggingArguments,
-    "warn,autopilot=debug,driver=debug,orderbook=debug,solver=debug,shared=debug"
+    "warn,autopilot=debug,driver=debug,orderbook=debug,solver=debug,shared=debug,cow_amm=debug"
 );
 
 #[derive(clap::Parser)]