Merge branch 'cowprotocol:main' into main

Author: xdecentralix

Cargo.lock

@@ -1,5 +1,6 @@
 //! A component that listens exclusively for `OrderRefund` events of the ethflow
 //! contract.
+
 use {
     ethcontract::{H160, H256, contract::AllEventsBuilder, transport::DynTransport},
     hex_literal::hex,

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

@@ -25,7 +25,7 @@ use {
     },
     hex_literal::hex,
     sqlx::{PgPool, types::BigDecimal},
-    std::{collections::HashMap, convert::TryInto, time::Duration},
+    std::{collections::HashMap, convert::TryInto},
     web3::types::U64,
 };
 
@@ -154,7 +154,7 @@ async fn settlement_deployment_block_number_hash(
     let block_number = deployment_block(GPv2Settlement::raw_contract(), chain_id)?;
     block_number_to_block_number_hash(web3, U64::from(block_number).into())
         .await
-        .ok_or_else(|| anyhow!("Deployment block not found"))
+        .context("Deployment block not found")
 }
 
 /// The block from which to start indexing eth-flow events. Note that this
@@ -241,17 +241,13 @@ pub async fn determine_ethflow_refund_indexing_start(
     .expect("Should be able to find a valid start block")
 }
 
-/// 1. Check the `last_indexed_blocks` table for the `index_name`. Use the next
-///    block as the starting point.
+/// 1. Check the `last_indexed_blocks` table for the `index_name`.
 /// 2. If no value found or the index is 0, use `fallback_start_block`, if
 ///    provided.
 /// 3. Fallback to the settlement deployment block number, if the `chain_id` is
 ///    provided.
 /// 4. Try to fetch the block number to ensure the node is able to continue
 ///    indexing.
-///
-/// Each option except the DB read is retried up to 3 times with a delay of
-/// 500ms.
 async fn find_indexing_start_block(
     db: &PgPool,
     web3: &Web3,
@@ -263,65 +259,28 @@ async fn find_indexing_start_block(
         .await
         .context("failed to read last indexed block from db")?;
 
-    let retries = 3;
-    let retry_delay = Duration::from_millis(500);
     if last_indexed_block > 0 {
-        return retry(
-            || async {
-                block_number_to_block_number_hash(web3, U64::from(last_indexed_block + 1).into())
-                    .await
-                    .map(Some)
-                    .context("failed to fetch block")
-            },
-            retries,
-            retry_delay,
-        )
-        .await;
+        return block_number_to_block_number_hash(web3, U64::from(last_indexed_block).into())
+            .await
+            .map(Some)
+            .context("failed to fetch block");
     }
     if let Some(start_block) = fallback_start_block {
-        return retry(
-            || async {
-                block_number_to_block_number_hash(web3, start_block.into())
-                    .await
-                    .map(Some)
-                    .context("failed to fetch fallback indexing start block")
-            },
-            retries,
-            retry_delay,
-        )
-        .await;
+        return block_number_to_block_number_hash(web3, start_block.into())
+            .await
+            .map(Some)
+            .context("failed to fetch fallback indexing start block");
     }
     if let Some(chain_id) = settlement_fallback_chain_id {
-        return retry(
-            || settlement_deployment_block_number_hash(web3, chain_id),
-            retries,
-            retry_delay,
-        )
-        .await
-        .map(Some)
-        .context("failed to fetch settlement deployment block");
+        return settlement_deployment_block_number_hash(web3, chain_id)
+            .await
+            .map(Some)
+            .context("failed to fetch settlement deployment block");
     }
 
     Ok(None)
 }
 
-async fn retry<F, T, E>(mut f: impl FnMut() -> F, retries: usize, delay: Duration) -> Result<T, E>
-where
-    F: Future<Output = Result<T, E>>,
-{
-    let mut attempts = 0;
-    loop {
-        match f().await {
-            Ok(val) => return Ok(val),
-            Err(_) if attempts < retries => {
-                attempts += 1;
-                tokio::time::sleep(delay).await;
-            }
-            Err(err) => return Err(err),
-        }
-    }
-}
-
 #[cfg(test)]
 mod test {
     use {

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

@@ -18,6 +18,7 @@ type SolutionId = u64;
 
 #[derive(Debug, Clone)]
 pub struct Solution {
+    /// A solution ID provided by the solver.
     id: SolutionId,
     solver: eth::Address,
     /// Score reported by the solver in their response.

crates/autopilot/src/domain/competition/mod.rs

@@ -26,6 +26,11 @@ impl Ranking {
         self.ranked.iter().chain(&self.filtered_out)
     }
 
+    /// Enumerates all solutions. The index is used as solution UID.
+    pub fn enumerated(&self) -> impl Iterator<Item = (usize, &Participant<Ranked>)> {
+        self.all().enumerate()
+    }
+
     /// All solutions that won the right to get executed.
     pub fn winners(&self) -> impl Iterator<Item = &Participant<Ranked>> {
         self.ranked.iter().filter(|p| p.is_winner())

crates/autopilot/src/domain/competition/winner_selection/mod.rs

@@ -325,6 +325,7 @@ impl From<infra::persistence::DatabaseError> for Error {
 pub struct ExecutionStarted {
     pub auction_id: AuctionId,
     pub solver: eth::Address,
+    pub solution_uid: usize,
     pub start_timestamp: DateTime<Utc>,
     pub start_block: u64,
     pub deadline_block: u64,
@@ -334,6 +335,7 @@ pub struct ExecutionStarted {
 pub struct ExecutionEnded {
     pub auction_id: AuctionId,
     pub solver: eth::Address,
+    pub solution_uid: usize,
     pub end_timestamp: DateTime<Utc>,
     pub end_block: u64,
     pub outcome: String,

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

@@ -823,6 +823,10 @@ impl Persistence {
             &mut ex,
             event.auction_id,
             ByteArray(event.solver.0.0),
+            event
+                .solution_uid
+                .try_into()
+                .context("solution uid overflow")?,
             event.start_timestamp,
             event
                 .start_block
@@ -852,6 +856,10 @@ impl Persistence {
             &mut ex,
             event.auction_id,
             ByteArray(event.solver.0.0),
+            event
+                .solution_uid
+                .try_into()
+                .context("solution uid overflow")?,
             event.end_timestamp,
             event.end_block.try_into().context("end block overflow")?,
             event.outcome,

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

@@ -388,7 +388,11 @@ pub async fn run(args: Arguments) {
         .unwrap();
 
     let skip_event_sync_start = if args.skip_event_sync {
-        block_number_to_block_number_hash(&web3, BlockNumber::Latest).await
+        Some(
+            block_number_to_block_number_hash(&web3, BlockNumber::Latest)
+                .await
+                .expect("Failed to fetch latest block"),
+        )
     } else {
         None
     };

crates/autopilot/src/run.rs

@@ -325,7 +325,10 @@ impl RunLoop {
             OrderEventLabel::Considered,
         );
 
-        for winner in ranking.winners() {
+        for (solution_uid, winner) in ranking
+            .enumerated()
+            .filter(|(_, participant)| participant.is_winner())
+        {
             let (driver, solution) = (winner.driver(), winner.solution());
             tracing::info!(driver = %driver.name, solution = %solution.id(), "winner");
 
@@ -334,6 +337,7 @@ impl RunLoop {
                 single_run_start,
                 driver,
                 solution,
+                solution_uid,
                 block_deadline,
             )
             .await;
@@ -349,6 +353,7 @@ impl RunLoop {
         single_run_start: Instant,
         driver: &Arc<infra::Driver>,
         solution: &Solution,
+        solution_uid: usize,
         block_deadline: u64,
     ) {
         let solved_order_uids: HashSet<_> = solution.orders().keys().cloned().collect();
@@ -369,10 +374,11 @@ impl RunLoop {
             match self_
                 .settle(
                     &driver_,
-                    solution_id,
                     solved_order_uids.clone(),
                     solver,
                     auction_id,
+                    solution_id,
+                    solution_uid,
                     block_deadline,
                 )
                 .await
@@ -457,8 +463,7 @@ impl RunLoop {
         }
 
         let mut solutions: Vec<_> = ranking
-            .all()
-            .enumerate()
+            .enumerated()
             .map(|(index, participant)| SolverSettlement {
                 solver: participant.driver().name.clone(),
                 solver_address: participant.solution().solver().0,
@@ -699,13 +704,15 @@ impl RunLoop {
 
     /// Execute the solver's solution. Returns Ok when the corresponding
     /// transaction has been mined.
+    #[allow(clippy::too_many_arguments)]
     async fn settle(
         &self,
         driver: &infra::Driver,
-        solution_id: u64,
         solved_order_uids: HashSet<OrderUid>,
         solver: eth::Address,
         auction_id: i64,
+        solution_id: u64,
+        solution_uid: usize,
         submission_deadline_latest_block: u64,
     ) -> Result<TxId, SettleError> {
         let settle = async move {
@@ -724,6 +731,7 @@ impl RunLoop {
             self.store_execution_started(
                 auction_id,
                 solver,
+                solution_uid,
                 current_block,
                 submission_deadline_latest_block,
             );
@@ -749,7 +757,7 @@ impl RunLoop {
             }
         };
 
-        self.store_execution_ended(solver, auction_id, &result);
+        self.store_execution_ended(solver, auction_id, solution_uid, &result);
 
         // Clean up the in-flight orders regardless the result.
         self.in_flight_orders
@@ -766,6 +774,7 @@ impl RunLoop {
         &self,
         auction_id: i64,
         solver: eth::Address,
+        solution_uid: usize,
         start_block: u64,
         deadline_block: u64,
     ) {
@@ -774,6 +783,7 @@ impl RunLoop {
             let execution_started = ExecutionStarted {
                 auction_id,
                 solver,
+                solution_uid,
                 start_timestamp: chrono::Utc::now(),
                 start_block,
                 deadline_block,
@@ -794,6 +804,7 @@ impl RunLoop {
         &self,
         solver: eth::Address,
         auction_id: i64,
+        solution_uid: usize,
         result: &Result<TxId, SettleError>,
     ) {
         let end_timestamp = chrono::Utc::now();
@@ -809,6 +820,7 @@ impl RunLoop {
             let execution_ended = ExecutionEnded {
                 auction_id,
                 solver,
+                solution_uid,
                 end_timestamp,
                 end_block: current_block,
                 outcome,

crates/autopilot/src/run_loop.rs

@@ -14,6 +14,7 @@ hex = { workspace = true }
 sqlx = { workspace = true }
 strum = { workspace = true }
 serde_json = { workspace = true }
+tracing = { workspace = true }
 
 [dev-dependencies]
 maplit = { workspace = true }