Post process settlements in parallel (#3853)

# Description
Currently we post-process settlements (i.e. associate the tx with a
solution proposed by a solver) serially. IIRC it was done that way
because it was simpler and that step simply ran in a background task at
the time.
Since then this logic has been moved onto the hot path so all the time
we spend there delays the creation of the next auction.
Additionally with the introduction of combinatorial auctions it's now
possible and relatively common to have multiple settlement transactions
in the same block. Whenever we have multiple settlements in one block
this PR should result in a significant performance uplift because we
process multiple concurrently.

# Changes
replaced loop to post-process auctions serially with logic that first
fetches ALL unprocessed settlements and then works on up to 10 of them
concurrently. This is fine for the post processing logic (as opposed to
the raw event indexing) because it's okay for post-processing to happen
out of order. So if the we successfully post-process settlement `n+1`
but not settlement `n` (e.g. due to a crash) the DB query would still
just return all unprocessed events instead of all NEW unprocessed
events.

I also adjusted how retrying works in this code. Instead of returning
some result enum (ok, invalid, nothing_to_do) to make the serial loop
retry something we now have a function `retry_with_sleep` that simply
retires a passed in future n times and returns an `Option` to indicate
whether it was successful or not.
This is used in 2 places:
1. fetching all outstanding settlements
2. post-processing individual outstanding settlement

With that the new logic should be as robust as the old one while being
IMO easier to reason about.

## How to test
Not sure how to test the improvement specifically. Since this is a
performance optimization and I don't really want to test the internals
of the implementation having a new test that makes sure that a big
number of settlements can be post-processed would be enough to test
correctness and for the performance aspect we'd have to deploy this on
the cluster.

Regarding the performance I temporarily deployed it to staging and it
produced the expected effect of reducing the spikiness that comes from
multiple settlements needing to be post-processed in the same block.
<img width="954" height="270" alt="Screenshot 2025-11-03 at 22 12 26"
src="https://github.com/user-attachments/assets/b459fb76-bf44-40db-894e-46ffd3d67f7d"
/>

Author: MartinquaXD

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

@@ -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/domain/settlement/observer.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/infra/persistence/mod.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/database/src/settlements.rs

@@ -31,17 +31,16 @@ pub struct SettlementEvent {
 }
 
 #[instrument(skip_all)]
-pub async fn get_settlement_without_auction(
+pub async fn get_settlements_without_auction(
     ex: &mut PgConnection,
-) -> Result<Option<SettlementEvent>, sqlx::Error> {
+) -> Result<Vec<SettlementEvent>, sqlx::Error> {
     const QUERY: &str = r#"
 SELECT block_number, log_index, tx_hash
 FROM settlements
 WHERE auction_id IS NULL
 ORDER BY block_number ASC
-LIMIT 1
     "#;
-    sqlx::query_as(QUERY).fetch_optional(ex).await
+    sqlx::query_as(QUERY).fetch_all(ex).await
 }
 
 #[instrument(skip_all)]
@@ -195,10 +194,8 @@ mod tests {
             .await
             .unwrap();
 
-        let settlement = get_settlement_without_auction(&mut db)
-            .await
-            .unwrap()
-            .unwrap();
+        let settlements = get_settlements_without_auction(&mut db).await.unwrap();
+        let settlement = &settlements[0];
 
         assert_eq!(settlement.block_number, event.block_number);
         assert_eq!(settlement.log_index, event.log_index);
@@ -207,8 +204,8 @@ mod tests {
             .await
             .unwrap();
 
-        let settlement = get_settlement_without_auction(&mut db).await.unwrap();
+        let settlement = get_settlements_without_auction(&mut db).await.unwrap();
 
-        assert!(settlement.is_none());
+        assert!(settlement.is_empty());
     }
 }