Merge branch 'main' into simulator-refactor

Author: m-sz

crates/autopilot/src/boundary/events/settlement.rs

@@ -1,5 +1,5 @@
 use {
-    crate::{database::Postgres, domain::settlement},
+    crate::database::Postgres,
     alloy::{
         primitives::Address,
         rpc::types::{Filter, Log},
@@ -35,16 +35,14 @@ impl AlloyEventRetrieving for GPv2SettlementContract {
 
 pub struct Indexer {
     db: Postgres,
-    start_index: u64,
-    settlement_observer: settlement::Observer,
+    start_indexing_block: u64,
 }
 
 impl Indexer {
-    pub fn new(db: Postgres, settlement_observer: settlement::Observer, start_index: u64) -> Self {
+    pub fn new(db: Postgres, start_indexing_block: u64) -> Self {
         Self {
             db,
-            settlement_observer,
-            start_index,
+            start_indexing_block,
         }
     }
 }
@@ -57,7 +55,7 @@ impl EventStoring<(GPv2SettlementEvents, Log)> for Indexer {
     async fn last_event_block(&self) -> Result<u64> {
         super::read_last_block_from_db(&self.db.pool, INDEX_NAME)
             .await
-            .map(|last_block| last_block.max(self.start_index))
+            .map(|last_block| last_block.max(self.start_indexing_block))
     }
 
     async fn persist_last_indexed_block(&mut self, latest_block: u64) -> Result<()> {
@@ -74,21 +72,13 @@ impl EventStoring<(GPv2SettlementEvents, Log)> for Indexer {
         crate::database::events::replace_events(&mut transaction, events, from_block).await?;
         database::settlements::delete(&mut transaction, from_block).await?;
         transaction.commit().await?;
-
-        self.settlement_observer
-            .post_process_outstanding_settlement_transactions()
-            .await;
         Ok(())
     }
 
     async fn append_events(&mut self, events: Vec<(GPv2SettlementEvents, Log)>) -> Result<()> {
         let mut transaction = self.db.pool.begin().await?;
         crate::database::events::append_events(&mut transaction, events).await?;
         transaction.commit().await?;
-
-        self.settlement_observer
-            .post_process_outstanding_settlement_transactions()
-            .await;
         Ok(())
     }
 }

crates/autopilot/src/maintenance.rs

@@ -3,17 +3,19 @@ use {
         boundary::events::settlement::{GPv2SettlementContract, Indexer},
         database::{
             Postgres,
+            ethflow_events::event_retriever::EthFlowRefundRetriever,
             onchain_order_events::{
                 OnchainOrderParser,
                 ethflow_events::{EthFlowData, EthFlowDataForDb},
                 event_retriever::CoWSwapOnchainOrdersContract,
             },
         },
+        domain::settlement,
         event_updater::EventUpdater,
     },
     anyhow::Result,
     ethrpc::block_stream::{BlockInfo, CurrentBlockWatcher, into_stream},
-    futures::StreamExt,
+    futures::{FutureExt, StreamExt},
     prometheus::{
         HistogramVec,
         IntCounterVec,
@@ -27,11 +29,13 @@ use {
     },
     tokio::sync::watch,
     tokio_stream::wrappers::WatchStream,
+    tracing::Instrument,
 };
 
 /// Component to sync with the maintenance logic that runs in a background task.
-/// This allows us to run the maintenance logic ASAP but still wait for it to
-/// finish in a convenient manner.
+/// This allows us to run the maintenance logic as soon as we see a new block
+/// while still making the autopilot run loop only wait for updates that are
+/// essential for building new auctions.
 #[derive(Clone)]
 pub struct MaintenanceSync {
     /// How long the autopilot wants to wait at most.
@@ -75,19 +79,26 @@ pub struct Maintenance {
     /// All indexing tasks to keep cow amms up to date.
     cow_amm_indexer: Vec<Arc<dyn Maintaining>>,
     /// Tasks to index ethflow orders that were submitted onchain.
-    ethflow_indexer: Vec<EthflowIndexer>,
+    ethflow_order_indexer: Vec<EthflowOrderIndexer>,
+    /// Tasks to index ethflow refunds.
+    ethflow_refund_indexer: Vec<EthflowRefundIndexer>,
+    /// Component to correctly attribute a settlement to a proposed solution.
+    settlement_observer: settlement::Observer,
 }
 
 impl Maintenance {
     pub fn new(
         settlement_indexer: EventUpdater<Indexer, GPv2SettlementContract>,
         db_cleanup: Postgres,
+        settlement_observer: settlement::Observer,
     ) -> Self {
         Self {
             settlement_indexer,
             db_cleanup,
             cow_amm_indexer: Default::default(),
-            ethflow_indexer: Default::default(),
+            ethflow_order_indexer: Default::default(),
+            ethflow_refund_indexer: Default::default(),
+            settlement_observer,
         }
     }
 
@@ -108,7 +119,12 @@ impl Maintenance {
                     .next()
                     .await
                     .expect("block stream terminated unexpectedly");
-                self.index_until_block(block, &sender).await;
+                self.index_until_block(block, &sender)
+                    .instrument(tracing::info_span!(
+                        "autopilot_maintenance",
+                        block = block.number
+                    ))
+                    .await;
             }
         });
 
@@ -122,33 +138,45 @@ impl Maintenance {
         metrics().last_seen_block.set(block.number);
         let start = Instant::now();
 
-        if let Err(err) = self.update_inner().await {
-            tracing::warn!(?err, block = block.number, "failed to run maintenance");
+        if let Err(err) = self.run_essential_maintenance().await {
+            tracing::warn!(?err, "failed to run essential maintenance");
             metrics().updates.with_label_values(&["error"]).inc();
             return;
         }
 
         tracing::info!(
-            block = block.number,
             time = ?start.elapsed(),
-            "successfully ran maintenance task"
+            "successfully ran essential maintenance tasks"
         );
         metrics().last_updated_block.set(block.number);
         metrics().updates.with_label_values(&["success"]).inc();
         if let Err(err) = last_processed_block.send(block.number) {
             tracing::warn!(?err, "nobody listening for processed blocks anymore");
         }
+
+        // only after we informed the run_loop that the essential updates are done we
+        // kick off the optional maintenance tasks
+        let start = Instant::now();
+        if let Err(err) = self.run_optional_maintenance().await {
+            tracing::warn!(?err, "failed to run optional maintenance");
+            return;
+        }
+        tracing::info!(
+            time = ?start.elapsed(),
+            "successfully ran optional maintenance tasks"
+        );
     }
 
-    async fn update_inner(&self) -> Result<()> {
-        let _timer =
-            observe::metrics::metrics().on_auction_overhead_start("autopilot", "maintenance_total");
+    /// Runs all the maintenance tasks that are needed to ensure the next
+    /// auction gets built using the most up-to-date information.
+    async fn run_essential_maintenance(&self) -> Result<()> {
+        let _timer = observe::metrics::metrics()
+            .on_auction_overhead_start("autopilot", "maintenance_essential");
         tokio::try_join!(
             Self::timed_future(
                 "settlement_indexer",
                 self.settlement_indexer.run_maintenance()
             ),
-            Self::timed_future("db_cleanup", self.db_cleanup.run_maintenance()),
             Self::timed_future(
                 "cow_amm_indexer",
                 futures::future::try_join_all(
@@ -158,9 +186,9 @@ impl Maintenance {
                 ),
             ),
             Self::timed_future(
-                "ethflow_indexer",
+                "ethflow_order_indexer",
                 futures::future::try_join_all(
-                    self.ethflow_indexer
+                    self.ethflow_order_indexer
                         .iter()
                         .map(|indexer| indexer.run_maintenance()),
                 ),
@@ -170,10 +198,41 @@ impl Maintenance {
         Ok(())
     }
 
+    /// Runs all the maintenance tasks that should run eventually but are not
+    /// very time sensitive.
+    async fn run_optional_maintenance(&self) -> Result<()> {
+        let _timer = observe::metrics::metrics()
+            .on_auction_overhead_start("autopilot", "maintenance_optional");
+        tokio::try_join!(
+            Self::timed_future("db_cleanup", self.db_cleanup.run_maintenance()),
+            Self::timed_future(
+                "ethflow_refund_indexer",
+                futures::future::try_join_all(
+                    self.ethflow_refund_indexer
+                        .iter()
+                        .map(|indexer| indexer.run_maintenance()),
+                ),
+            ),
+            Self::timed_future(
+                "settlement_attribution",
+                self.settlement_observer
+                    .post_process_outstanding_settlement_transactions()
+                    .map(|_| Ok(()))
+            )
+        )?;
+
+        Ok(())
+    }
+
     /// Registers all maintenance tasks that are necessary to correctly support
     /// ethflow orders.
-    pub fn add_ethflow_indexer(&mut self, ethflow_indexer: EthflowIndexer) {
-        self.ethflow_indexer.push(ethflow_indexer);
+    pub fn add_ethflow_indexing(
+        &mut self,
+        order_indexer: EthflowOrderIndexer,
+        refund_indexer: EthflowRefundIndexer,
+    ) {
+        self.ethflow_order_indexer.push(order_indexer);
+        self.ethflow_refund_indexer.push(refund_indexer);
     }
 
     /// Registers all maintenance tasks that are necessary to correctly support
@@ -194,9 +253,11 @@ impl Maintenance {
     }
 }
 
-type EthflowIndexer =
+type EthflowOrderIndexer =
     EventUpdater<OnchainOrderParser<EthFlowData, EthFlowDataForDb>, CoWSwapOnchainOrdersContract>;
 
+type EthflowRefundIndexer = EventUpdater<Postgres, EthFlowRefundRetriever>;
+
 #[derive(prometheus_metric_storage::MetricStorage)]
 #[metric(subsystem = "autopilot_maintenance")]
 struct Metrics {

crates/autopilot/src/run.rs

@@ -46,7 +46,6 @@ use {
         baseline_solver::BaseTokens,
         code_fetching::CachedCodeFetcher,
         http_client::HttpClientFactory,
-        maintenance::ServiceMaintenance,
         order_quoting::{self, OrderQuoter},
         price_estimation::factory::{self, PriceEstimatorFactory},
         signature_validator,
@@ -426,8 +425,6 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
         infra::persistence::Persistence::new(args.s3.into().unwrap(), Arc::new(db_write.clone()))
             .instrument(info_span!("persistence_init"))
             .await;
-    let settlement_observer =
-        crate::domain::settlement::Observer::new(eth.clone(), persistence.clone());
     let settlement_contract_start_index = match GPv2Settlement::deployment_block(&chain_id) {
         Some(block) => {
             tracing::debug!(block, "found settlement contract deployment");
@@ -449,7 +446,6 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
         ),
         boundary::events::settlement::Indexer::new(
             db_write.clone(),
-            settlement_observer,
             settlement_contract_start_index,
         ),
         block_retriever.clone(),
@@ -568,8 +564,14 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
     // updated in background task
     let trusted_tokens =
         AutoUpdatingTokenList::from_configuration(market_makable_token_list_configuration).await;
+    let settlement_observer =
+        crate::domain::settlement::Observer::new(eth.clone(), persistence.clone());
 
-    let mut maintenance = Maintenance::new(settlement_event_indexer, db_write.clone());
+    let mut maintenance = Maintenance::new(
+        settlement_event_indexer,
+        db_write.clone(),
+        settlement_observer,
+    );
     maintenance.add_cow_amm_indexer(&cow_amm_registry);
 
     if !args.ethflow_contracts.is_empty() {
@@ -626,13 +628,7 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
         .await
         .expect("Should be able to initialize event updater. Database read issues?");
 
-        maintenance.add_ethflow_indexer(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()),
-        );
+        maintenance.add_ethflow_indexing(onchain_order_indexer, refund_event_handler);
     }
 
     let run_loop_config = run_loop::Config {