Fetch solver that created settlement transaction (#3003)

# Description
A preparation to implement
#2998 which will require
sending multiple settle requests to drivers, and waiting for multiple
settlement transactions.

In order to match settle requests with settlement transactions, we need
`solver` that created the settlement transaction.

# Changes
<!-- List of detailed changes (how the change is accomplished) -->

- [ ] Fetch solver from settlement event and use it to make sure it's
the same as the driver reported solver address.

## How to test
Existing e2e tests.

crates/autopilot/src/database.rs

@@ -13,7 +13,6 @@ pub mod fee_policies;
 pub mod onchain_order_events;
 pub mod order_events;
 mod quotes;
-pub mod recent_settlements;
 
 #[derive(Debug, Clone)]
 pub struct Config {

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

@@ -179,20 +179,26 @@ impl Persistence {
         ex.commit().await.context("commit")
     }
 
-    /// For a given auction, finds all settlements and returns their transaction
-    /// hashes.
-    pub async fn find_settlement_transactions(
+    /// For a given auction and solver, tries to find the settlement
+    /// transaction.
+    pub async fn find_settlement_transaction(
         &self,
         auction_id: i64,
-    ) -> Result<Vec<eth::TxId>, DatabaseError> {
-        Ok(self
-            .postgres
-            .find_settlement_transactions(auction_id)
-            .await
-            .map_err(DatabaseError)?
-            .into_iter()
-            .map(eth::TxId)
-            .collect())
+        solver: eth::Address,
+    ) -> Result<Option<eth::TxId>, DatabaseError> {
+        let _timer = Metrics::get()
+            .database_queries
+            .with_label_values(&["find_settlement_transaction"])
+            .start_timer();
+
+        let mut ex = self.postgres.pool.acquire().await.context("acquire")?;
+        Ok(database::settlements::find_settlement_transaction(
+            &mut ex,
+            auction_id,
+            ByteArray(solver.0 .0),
+        )
+        .await?
+        .map(|hash| H256(hash.0).into()))
     }
 
     /// Checks if an auction already has an accociated settlement.

crates/autopilot/src/run_loop.rs

@@ -286,6 +286,7 @@ impl RunLoop {
             .extend(solved_order_uids.clone());
 
         let solution_id = solution.id();
+        let solver = solution.solver();
         let self_ = self.clone();
         let driver_ = driver.clone();
 
@@ -298,14 +299,15 @@ impl RunLoop {
                     &driver_,
                     solution_id,
                     solved_order_uids,
+                    solver,
                     auction_id,
                     block_deadline,
                 )
                 .await
             {
                 Ok(tx_hash) => {
                     Metrics::settle_ok(&driver_, submission_start.elapsed());
-                    tracing::debug!(?tx_hash, driver = %driver_.name, "solution settled");
+                    tracing::debug!(?tx_hash, driver = %driver_.name, ?solver, "solution settled");
                 }
                 Err(err) => {
                     Metrics::settle_err(&driver_, submission_start.elapsed(), &err);
@@ -751,14 +753,35 @@ impl RunLoop {
         driver: &infra::Driver,
         solution_id: u64,
         solved_order_uids: HashSet<OrderUid>,
+        solver: eth::Address,
         auction_id: i64,
         submission_deadline_latest_block: u64,
     ) -> Result<TxId, SettleError> {
         let request = settle::Request {
             solution_id,
             submission_deadline_latest_block,
         };
-        let result = self.wait_for_settlement(driver, auction_id, request).await;
+
+        // Wait for either the settlement transaction to be mined or the driver returned
+        // a result.
+        let result = match futures::future::select(
+            Box::pin(self.wait_for_settlement_transaction(
+                auction_id,
+                self.config.submission_deadline,
+                solver,
+            )),
+            Box::pin(driver.settle(&request, self.config.max_settlement_transaction_wait)),
+        )
+        .await
+        {
+            futures::future::Either::Left((res, _)) => res,
+            futures::future::Either::Right((driver_result, wait_for_settlement_transaction)) => {
+                match driver_result {
+                    Ok(_) => wait_for_settlement_transaction.await,
+                    Err(err) => Err(SettleError::Failure(err)),
+                }
+            }
+        };
 
         // Clean up the in-flight orders regardless the result.
         self.in_flight_orders
@@ -769,41 +792,16 @@ impl RunLoop {
         result
     }
 
-    /// Wait for either the settlement transaction to be mined or the driver
-    /// returned a result.
-    async fn wait_for_settlement(
-        &self,
-        driver: &infra::Driver,
-        auction_id: i64,
-        request: settle::Request,
-    ) -> Result<eth::TxId, SettleError> {
-        match futures::future::select(
-            Box::pin(
-                self.wait_for_settlement_transaction(auction_id, self.config.submission_deadline),
-            ),
-            Box::pin(driver.settle(&request, self.config.max_settlement_transaction_wait)),
-        )
-        .await
-        {
-            futures::future::Either::Left((res, _)) => res,
-            futures::future::Either::Right((driver_result, onchain_task)) => {
-                driver_result.map_err(|err| {
-                    tracing::warn!(?err, "driver settle request failed");
-                    SettleError::Failure(err)
-                })?;
-                onchain_task.await
-            }
-        }
-    }
-
-    /// Tries to find a `settle` contract call with calldata ending in `tag`.
+    /// Tries to find a `settle` contract call with calldata ending in `tag` and
+    /// originated from the `solver`.
     ///
     /// Returns None if no transaction was found within the deadline or the task
     /// is cancelled.
     async fn wait_for_settlement_transaction(
         &self,
         auction_id: i64,
         max_blocks_wait: u64,
+        solver: eth::Address,
     ) -> Result<eth::TxId, SettleError> {
         let current = self.eth.current_block().borrow().number;
         let deadline = current.saturating_add(max_blocks_wait);
@@ -816,17 +814,18 @@ impl RunLoop {
 
             match self
                 .persistence
-                .find_settlement_transactions(auction_id)
+                .find_settlement_transaction(auction_id, solver)
                 .await
             {
-                Ok(hashes) if hashes.is_empty() => {}
-                Ok(hashes) => {
-                    if let Some(hash) = hashes.into_iter().next() {
-                        return Ok(hash);
-                    }
-                }
+                Ok(Some(transaction)) => return Ok(transaction),
+                Ok(None) => {}
                 Err(err) => {
-                    tracing::warn!(?err, "failed to fetch recent settlement tx hashes");
+                    tracing::warn!(
+                        ?err,
+                        ?auction_id,
+                        ?solver,
+                        "failed to find settlement transaction"
+                    );
                 }
             }
             if block.number >= deadline {

crates/database/src/settlements.rs

@@ -1,37 +1,24 @@
 use {
-    crate::{events::EventIndex, PgTransaction, TransactionHash},
+    crate::{Address, PgTransaction, TransactionHash},
     sqlx::{Executor, PgConnection},
 };
 
-pub async fn get_hash_by_event(
-    ex: &mut PgConnection,
-    event: &EventIndex,
-) -> Result<TransactionHash, sqlx::Error> {
-    const QUERY: &str = r#"
-SELECT tx_hash
-FROM settlements
-WHERE
-    block_number = $1 AND
-    log_index = $2
-    "#;
-    sqlx::query_scalar::<_, TransactionHash>(QUERY)
-        .bind(event.block_number)
-        .bind(event.log_index)
-        .fetch_one(ex)
-        .await
-}
-
-pub async fn get_hashes_by_auction_id(
+pub async fn find_settlement_transaction(
     ex: &mut PgConnection,
     auction_id: i64,
-) -> Result<Vec<TransactionHash>, sqlx::Error> {
+    solver: Address,
+) -> Result<Option<TransactionHash>, sqlx::Error> {
     const QUERY: &str = r#"
 SELECT tx_hash
 FROM settlements
 WHERE
-    auction_id = $1
+    auction_id = $1 AND solver = $2
     "#;
-    sqlx::query_as(QUERY).bind(auction_id).fetch_all(ex).await
+    sqlx::query_as(QUERY)
+        .bind(auction_id)
+        .bind(solver)
+        .fetch_optional(ex)
+        .await
 }
 
 #[derive(Debug, sqlx::FromRow)]