Merge branch 'main' into jemalloc-profiling-support

Author: squadgazzz

Cargo.lock

@@ -101,6 +101,13 @@ tower-http = "0.4"
 tracing-opentelemetry = "0.31"
 tracing-serde = "0.2"
 vergen = "8"
+walkdir = "2.5.0"
+quote = "1.0.41"
+syn = "2.0.108"
+prettyplease = "0.2.37"
+proc-macro2 = "1.0.103"
+alloy-sol-macro-input = "1.4.1"
+alloy-sol-macro-expander = "1.4.1"
 
 [workspace.lints]
 clippy.cast_possible_wrap = "deny"

Cargo.toml

@@ -28,7 +28,7 @@ impl AlloyEventRetrieving for GPv2SettlementContract {
         Filter::new().address(self.address)
     }
 
-    fn provider(&self) -> &contracts::alloy::Provider {
+    fn provider(&self) -> &alloy::providers::DynProvider {
         &self.provider
     }
 }
@@ -75,7 +75,9 @@ impl EventStoring<(GPv2SettlementEvents, Log)> for Indexer {
         database::settlements::delete(&mut transaction, from_block).await?;
         transaction.commit().await?;
 
-        self.settlement_observer.update().await;
+        self.settlement_observer
+            .post_process_outstanding_settlement_transactions()
+            .await;
         Ok(())
     }
 
@@ -84,7 +86,9 @@ impl EventStoring<(GPv2SettlementEvents, Log)> for Indexer {
         crate::database::events::append_events(&mut transaction, events).await?;
         transaction.commit().await?;
 
-        self.settlement_observer.update().await;
+        self.settlement_observer
+            .post_process_outstanding_settlement_transactions()
+            .await;
         Ok(())
     }
 }

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

@@ -29,7 +29,7 @@ impl EthFlowRefundRetriever {
 impl AlloyEventRetrieving for EthFlowRefundRetriever {
     type Event = CoWSwapEthFlow::CoWSwapEthFlowEvents;
 
-    fn provider(&self) -> &contracts::alloy::Provider {
+    fn provider(&self) -> &alloy::providers::DynProvider {
         &self.web3.alloy
     }
 

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

@@ -42,7 +42,7 @@ impl AlloyEventRetrieving for CoWSwapOnchainOrdersContract {
             ]))
     }
 
-    fn provider(&self) -> &contracts::alloy::Provider {
+    fn provider(&self) -> &alloy::providers::DynProvider {
         &self.web3.alloy
     }
 }

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

@@ -600,9 +600,9 @@ fn convert_onchain_order_placement(
     // executed fast (we don't want to reserve the user's ETH for too long)
     if quote.as_ref().is_ok_and(|quote| {
         !order_data.within_market(QuoteAmounts {
-            sell: quote.sell_amount.into_legacy(),
-            buy: quote.buy_amount.into_legacy(),
-            fee: quote.fee_amount.into_legacy(),
+            sell: quote.sell_amount,
+            buy: quote.buy_amount,
+            fee: quote.fee_amount,
         })
     }) {
         tracing::debug!(%order_uid, ?owner, "order is outside market price");
@@ -786,7 +786,7 @@ mod test {
         super::*,
         crate::database::Config,
         alloy::primitives::U256,
-        contracts::alloy::{CoWSwapOnchainOrders, InstanceExt},
+        contracts::alloy::CoWSwapOnchainOrders,
         database::{byte_array::ByteArray, onchain_broadcasted_orders::OnchainOrderPlacement},
         ethcontract::H160,
         ethrpc::Web3,

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

@@ -20,6 +20,8 @@ use {
     },
     anyhow::{Context, Result, anyhow},
     ethrpc::alloy::conversions::IntoLegacy,
+    futures::StreamExt,
+    rand::Rng,
     std::time::Duration,
 };
 
@@ -29,80 +31,68 @@ pub struct Observer {
     persistence: infra::Persistence,
 }
 
-enum IndexSuccess {
-    NothingToDo,
-    IndexedSettlement,
-    SkippedInvalidTransaction,
-}
-
 impl Observer {
     /// Creates a new Observer and asynchronously schedules the first update
     /// run.
     pub fn new(eth: infra::Ethereum, persistence: infra::Persistence) -> Self {
         Self { eth, persistence }
     }
 
-    /// Fetches all the available missing data needed for bookkeeping.
-    /// This needs to get called after indexing a new settlement event
-    /// since this code needs that data to already be present in the DB.
-    pub async fn update(&self) {
-        const MAX_RETRIES: usize = 5;
-        let mut attempts = 0;
-        while attempts < MAX_RETRIES {
-            match self.single_update().await {
-                Ok(IndexSuccess::IndexedSettlement) => {
-                    tracing::debug!("on settlement event updater ran and processed event");
-                }
-                Ok(IndexSuccess::SkippedInvalidTransaction) => {
-                    tracing::warn!("stored default values for unindexable transaction");
-                }
-                Ok(IndexSuccess::NothingToDo) => {
-                    tracing::debug!("on settlement event updater ran without update");
+    /// Post processes all outstanding settlements. This involves decoding the
+    /// settlement details from the transaction and associating it with a
+    /// solution proposed by a solver for the auction specified at the end of
+    /// the transaction call data. If no solution can be found a dummy mapping
+    /// gets saved to mark the settlement as processed. This can happen when a
+    /// solver submits a solution despite not winning or if the settlement
+    /// belongs to an auction that was arbitrated in another environment (i.e.
+    /// prod vs. staging).
+    pub async fn post_process_outstanding_settlement_transactions(&self) {
+        let settlements =
+            match Self::retry_with_sleep(|| self.persistence.get_settlements_without_auction())
+                .await
+            {
+                Ok(settlements) => settlements,
+                Err(errs) => {
+                    tracing::warn!(?errs, "failed to fetch unprocessed settlements");
                     return;
                 }
-                Err(err) => {
-                    tracing::debug!(?err, "encountered retryable error");
-                    // wait a little to give temporary errors a chance to resolve themselves
-                    const TEMP_ERROR_BACK_OFF: Duration = Duration::from_millis(100);
-                    tokio::time::sleep(TEMP_ERROR_BACK_OFF).await;
-                    attempts += 1;
-                    continue;
-                }
-            }
+            };
 
-            // everything worked fine -> reset our attempts for the next settlement
-            attempts = 0;
+        if settlements.is_empty() {
+            tracing::debug!("no unprocessed settlements found");
+            return;
         }
-    }
-
-    /// Update database for settlement events that have not been processed yet.
-    ///
-    /// Returns whether an update was performed.
-    async fn single_update(&self) -> Result<IndexSuccess> {
-        // Find a settlement event that has not been processed yet.
-        let Some(event) = self
-            .persistence
-            .get_settlement_without_auction()
-            .await
-            .context("failed to fetch unprocessed tx from DB")?
-        else {
-            return Ok(IndexSuccess::NothingToDo);
-        };
 
-        tracing::debug!(tx = ?event.transaction, "found unprocessed settlement");
+        // On mainnet it's common to have multiple settlements in the
+        // same block. So even if we process every block immediately,
+        // we should still post-process multiple settlements concurrently.
+        const MAX_CONCURRENCY: usize = 10;
+        futures::stream::iter(settlements)
+            .for_each_concurrent(MAX_CONCURRENCY, |settlement| async move {
+                tracing::debug!(tx = ?settlement.transaction, "start post processing of settlement");
+                match Self::retry_with_sleep(|| self.post_process_settlement(settlement)).await {
+                    Ok(_) =>  tracing::debug!(
+                        tx = ?settlement.transaction,
+                        "successfully post-processed settlement"
+                    ),
+                    Err(errs) => tracing::warn!(
+                        tx = ?settlement.transaction,
+                        ?errs,
+                        "gave up on post-processing settlement"
+                    ),
+                }
+            })
+            .await;
+    }
 
+    async fn post_process_settlement(&self, settlement: eth::SettlementEvent) -> Result<()> {
         let settlement_data = self
-            .fetch_auction_data_for_transaction(event.transaction)
+            .fetch_auction_data_for_transaction(settlement.transaction)
             .await?;
         self.persistence
-            .save_settlement(event, settlement_data.as_ref())
+            .save_settlement(settlement, settlement_data.as_ref())
             .await
-            .context("failed to update settlement")?;
-
-        match settlement_data {
-            None => Ok(IndexSuccess::SkippedInvalidTransaction),
-            Some(_) => Ok(IndexSuccess::IndexedSettlement),
-        }
+            .context("failed to update settlement")
     }
 
     /// Inspects the calldata of the transaction, decodes the arguments, and
@@ -179,4 +169,28 @@ impl Observer {
             }
         }
     }
+
+    async fn retry_with_sleep<F, OK, ERR>(future: impl Fn() -> F) -> Result<OK, Vec<ERR>>
+    where
+        F: Future<Output = Result<OK, ERR>>,
+        ERR: std::fmt::Debug,
+    {
+        const MAX_RETRIES: usize = 5;
+
+        let mut errors = Vec::new();
+        let mut tries = 0;
+        while tries < MAX_RETRIES {
+            match future().await {
+                Ok(res) => return Ok(res),
+                Err(err) => {
+                    errors.push(err);
+                    tries += 1;
+                    // wait a little to give temporary errors a chance to resolve themselves
+                    let timeout_with_jitter = 50u64 + rand::thread_rng().gen_range(0..=50);
+                    tokio::time::sleep(Duration::from_millis(timeout_with_jitter)).await;
+                }
+            }
+        }
+        Err(errors)
+    }
 }

crates/autopilot/src/domain/settlement/observer.rs

@@ -6,7 +6,6 @@ use {
         GPv2AllowListAuthentication,
         GPv2Settlement,
         HooksTrampoline,
-        InstanceExt,
         WETH9,
         support::Balances,
     },

crates/autopilot/src/infra/blockchain/contracts.rs

@@ -594,17 +594,18 @@ impl Persistence {
 
     /// Returns the oldest settlement event for which the accociated auction is
     /// not yet populated in the database.
-    pub async fn get_settlement_without_auction(
+    pub async fn get_settlements_without_auction(
         &self,
-    ) -> Result<Option<domain::eth::SettlementEvent>, DatabaseError> {
+    ) -> Result<Vec<domain::eth::SettlementEvent>, DatabaseError> {
         let _timer = Metrics::get()
             .database_queries
             .with_label_values(&["get_settlement_without_auction"])
             .start_timer();
 
         let mut ex = self.postgres.pool.acquire().await?;
-        let event = database::settlements::get_settlement_without_auction(&mut ex)
+        let events = database::settlements::get_settlements_without_auction(&mut ex)
             .await?
+            .into_iter()
             .map(|event| {
                 let event = domain::eth::SettlementEvent {
                     block: u64::try_from(event.block_number)
@@ -615,8 +616,8 @@ impl Persistence {
                 };
                 Ok::<_, DatabaseError>(event)
             })
-            .transpose()?;
-        Ok(event)
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(events)
     }
 
     /// Returns the trade events that are associated with the settlement event

crates/autopilot/src/infra/persistence/mod.rs

@@ -27,7 +27,7 @@ use {
     alloy::eips::BlockNumberOrTag,
     chain::Chain,
     clap::Parser,
-    contracts::alloy::{BalancerV2Vault, GPv2Settlement, IUniswapV3Factory, InstanceExt, WETH9},
+    contracts::alloy::{BalancerV2Vault, GPv2Settlement, IUniswapV3Factory, WETH9},
     ethcontract::H160,
     ethrpc::{
         Web3,