Move refund indexing back into background task (#2935)

# Description
Indexing refunds is not strictly necessary to have all relevant state
for building the next auction. That's why it's okay to move that logic
off of the critical path.
This should resolve the start up issues we've been seeing on arbitrum
staging where we have to reindex lots of blocks because the blocktime is
so fast and we don't do refunds regularly. However, we should still find
a better way to store the blocks we already indexed to avoid these
issues in general.

# Changes
Index refund events in a separate background task again.

crates/autopilot/src/maintenance.rs

@@ -3,7 +3,6 @@ use {
         arguments::RunLoopMode,
         boundary::events::settlement::{GPv2SettlementContract, Indexer},
         database::{
-            ethflow_events::event_retriever::EthFlowRefundRetriever,
             onchain_order_events::{
                 ethflow_events::{EthFlowData, EthFlowDataForDb},
                 event_retriever::CoWSwapOnchainOrdersContract,
@@ -36,8 +35,6 @@ pub struct Maintenance {
     settlement_indexer: EventUpdater<Indexer, GPv2SettlementContract>,
     /// Indexes ethflow orders (orders selling native ETH).
     ethflow_indexer: Option<EthflowIndexer>,
-    /// Indexes refunds issued for unsettled ethflow orders.
-    refund_indexer: Option<EventUpdater<Postgres, EthFlowRefundRetriever>>,
     /// Used for periodic cleanup tasks to not have the DB overflow with old
     /// data.
     db_cleanup: Postgres,
@@ -58,7 +55,6 @@ impl Maintenance {
             settlement_indexer,
             db_cleanup,
             cow_amm_indexer: Default::default(),
-            refund_indexer: None,
             ethflow_indexer: None,
             last_processed: Default::default(),
         }
@@ -98,7 +94,6 @@ impl Maintenance {
         tokio::try_join!(
             self.settlement_indexer.run_maintenance(),
             self.db_cleanup.run_maintenance(),
-            self.index_refunds(),
             self.index_ethflow_orders(),
             futures::future::try_join_all(
                 self.cow_amm_indexer
@@ -113,26 +108,14 @@ impl Maintenance {
 
     /// Registers all maintenance tasks that are necessary to correctly support
     /// ethflow orders.
-    pub fn with_ethflow(
-        &mut self,
-        ethflow_indexer: EthflowIndexer,
-        refund_indexer: EventUpdater<Postgres, EthFlowRefundRetriever>,
-    ) {
+    pub fn with_ethflow(&mut self, ethflow_indexer: EthflowIndexer) {
         self.ethflow_indexer = Some(ethflow_indexer);
-        self.refund_indexer = Some(refund_indexer);
     }
 
     pub fn with_cow_amms(&mut self, registry: &cow_amm::Registry) {
         self.cow_amm_indexer = registry.maintenance_tasks().clone();
     }
 
-    async fn index_refunds(&self) -> Result<()> {
-        if let Some(indexer) = &self.refund_indexer {
-            return indexer.run_maintenance().await;
-        }
-        Ok(())
-    }
-
     async fn index_ethflow_orders(&self) -> Result<()> {
         if let Some(indexer) = &self.ethflow_indexer {
             return indexer.run_maintenance().await;

crates/autopilot/src/run.rs

@@ -41,6 +41,7 @@ use {
         baseline_solver::BaseTokens,
         code_fetching::CachedCodeFetcher,
         http_client::HttpClientFactory,
+        maintenance::ServiceMaintenance,
         metrics::LivenessChecking,
         order_quoting::{self, OrderQuoter},
         price_estimation::factory::{self, PriceEstimatorFactory},
@@ -510,7 +511,13 @@ pub async fn run(args: Arguments) {
         .await
         .expect("Should be able to initialize event updater. Database read issues?");
 
-        maintenance.with_ethflow(onchain_order_indexer, refund_event_handler);
+        maintenance.with_ethflow(onchain_order_indexer);
+        // refunds are not critical for correctness and can therefore be indexed
+        // sporadically in a background task
+        let service_maintainer = ServiceMaintenance::new(vec![Arc::new(refund_event_handler)]);
+        tokio::task::spawn(
+            service_maintainer.run_maintenance_on_new_block(eth.current_block().clone()),
+        );
     }
 
     let run = RunLoop {