From a0e01067d069c27449ed3b31b1560509907f9a6f Mon Sep 17 00:00:00 2001 From: MartinquaXD Date: Thu, 3 Sep 2026 10:18:34 +0000 Subject: [PATCH] Extract SettleCallCoordinator from run_loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moves `settle`, `wait_for_settlement_transaction`, `store_execution_started/ended`, and the `SettleError` enum out of `RunLoop` into a new `run_loop::settle_call_coordinator` submodule. `RunLoop` holds an `Arc` built once in `new`; `start_settlement_execution` calls `self.settle_coordinator.settle(...)` with the same `settle::Request` it always built. Pure refactor — no behavior change — so the fast-path handler in a later PR can reuse the submission machinery without ballooning that PR. Signed-off-by: MartinquaXD --- crates/autopilot/src/run_loop.rs | 220 ++---------------- .../src/run_loop/settle_call_coordinator.rs | 217 +++++++++++++++++ 2 files changed, 241 insertions(+), 196 deletions(-) create mode 100644 crates/autopilot/src/run_loop/settle_call_coordinator.rs diff --git a/crates/autopilot/src/run_loop.rs b/crates/autopilot/src/run_loop.rs index 682b6e682e..1e18d98a7d 100644 --- a/crates/autopilot/src/run_loop.rs +++ b/crates/autopilot/src/run_loop.rs @@ -1,4 +1,7 @@ +pub mod settle_call_coordinator; + use { + self::settle_call_coordinator::{SettleCallCoordinator, SettleError}, crate::{ domain::{ self, @@ -9,7 +12,6 @@ use { Unscored, winner_selection::{self, Ranking}, }, - settlement::{ExecutionEnded, ExecutionStarted}, }, infra::{ self, @@ -27,9 +29,9 @@ use { anyhow::{Context, Result}, chrono::{DateTime, Utc}, database::order_events::OrderEventLabel, - eth_domain_types::{self as eth, Address, TxId}, + eth_domain_types::Address, ethrpc::block_stream::{BlockInfo, CurrentBlockWatcher}, - futures::{FutureExt, StreamExt, TryFutureExt}, + futures::{StreamExt, TryFutureExt}, itertools::Itertools, num::ToPrimitive, rand::seq::SliceRandom, @@ -166,6 +168,9 @@ pub struct RunLoop { /// Drivers that do NOT support delta auctions drivers: Vec>, + /// Sends `/settle` calls to drivers and waits for the resulting + /// transaction to be mined. + settle_coordinator: Arc, } impl RunLoop { @@ -186,6 +191,13 @@ impl RunLoop { Self::spawn_block_listener(eth.current_block().clone(), wake_runloop.clone()); + let settle_coordinator = Arc::new(SettleCallCoordinator::new( + eth.clone(), + persistence.clone(), + maintenance.clone(), + config.max_settlement_transaction_wait, + )); + Self { delta_state: std::sync::Mutex::new(DeltaState::new( config.auction_delta_checkpoint_interval, @@ -200,6 +212,7 @@ impl RunLoop { winner_selection: winner_selection::Arbitrator::new(max_winners, weth), wake_notify: wake_runloop, drivers, + settle_coordinator, } } @@ -487,15 +500,15 @@ impl RunLoop { tracing::info!(driver = %driver_.name, solution = %solution_id, "settling"); let submission_start = Instant::now(); + let request = settle::Request { + solution_id, + submission_deadline_latest_block: block_deadline, + auction_id, + }; + match self_ - .settle( - &driver_, - solver, - auction_id, - solution_id, - solution_uid, - block_deadline, - ) + .settle_coordinator + .settle(&driver_, solver, solution_uid, request) .await { Ok(tx_hash) => { @@ -768,183 +781,6 @@ impl RunLoop { } Ok(response.into_domain()) } - - /// Execute the solver's solution. Returns Ok when the corresponding - /// transaction has been mined. - #[instrument(skip_all, fields(driver = driver.name, solution_uid))] - async fn settle( - &self, - driver: &infra::Driver, - solver: eth::Address, - auction_id: i64, - solution_id: u64, - solution_uid: usize, - submission_deadline_latest_block: u64, - ) -> Result { - let settle = async move { - let current_block = self.eth.current_block().borrow().number; - anyhow::ensure!( - current_block < submission_deadline_latest_block, - "submission deadline was missed" - ); - - let request = settle::Request { - solution_id, - submission_deadline_latest_block, - auction_id, - }; - - self.store_execution_started( - auction_id, - solver, - solution_uid, - current_block, - submission_deadline_latest_block, - ); - driver - .settle(&request, self.config.max_settlement_transaction_wait) - .await - } - .boxed(); - - let wait_for_settlement_transaction = self - .wait_for_settlement_transaction( - auction_id, - solver, - submission_deadline_latest_block, - solution_uid, - ) - .boxed(); - - // Wait for either the settlement transaction to be mined or the driver - // returned a result. - let result = match futures::future::select(wait_for_settlement_transaction, settle).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::Other(err)), - } - } - }; - - self.store_execution_ended(solver, auction_id, solution_uid, &result); - - result - } - - /// Stores settlement execution started event in the DB in a background task - /// to not block the runloop. - fn store_execution_started( - &self, - auction_id: i64, - solver: eth::Address, - solution_uid: usize, - start_block: u64, - deadline_block: u64, - ) { - let persistence = self.persistence.clone(); - tokio::spawn(async move { - let execution_started = ExecutionStarted { - auction_id, - solver, - solution_uid, - start_timestamp: chrono::Utc::now(), - start_block, - deadline_block, - }; - - if let Err(err) = persistence - .store_settlement_execution_started(execution_started) - .await - { - tracing::error!(?err, "failed to store settlement execution event"); - } - }); - } - - /// Stores settlement execution ended event in the DB in a background task - /// to not block the runloop. - fn store_execution_ended( - &self, - solver: eth::Address, - auction_id: i64, - solution_uid: usize, - result: &Result, - ) { - let end_timestamp = chrono::Utc::now(); - let current_block = self.eth.current_block().borrow().number; - let persistence = self.persistence.clone(); - let outcome = match result { - Ok(_) => "success".to_string(), - Err(SettleError::Timeout) => "timeout".to_string(), - Err(SettleError::Other(err)) => format!("driver failed: {err}"), - }; - - tokio::spawn(async move { - let execution_ended = ExecutionEnded { - auction_id, - solver, - solution_uid, - end_timestamp, - end_block: current_block, - outcome, - }; - if let Err(err) = persistence - .store_settlement_execution_ended(execution_ended) - .await - { - tracing::error!(?err, "failed to update settlement execution event"); - } - }); - } - - /// 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. - #[instrument(skip_all)] - async fn wait_for_settlement_transaction( - &self, - auction_id: i64, - solver: eth::Address, - submission_deadline_latest_block: u64, - solution_uid: usize, - ) -> Result { - let current = self.eth.current_block().borrow().number; - tracing::debug!(%current, deadline=%submission_deadline_latest_block, %auction_id, "waiting for tag"); - loop { - let block = ethrpc::block_stream::next_block(self.eth.current_block()).await; - // Run maintenance to ensure the system processed the last available - // block so it's possible to find the tx in the DB in - // the next line. - self.maintenance - .wait_until_block_processed(SyncTarget::FullyProcessed(block.number)) - .await; - - match self - .persistence - .find_settlement_transaction(auction_id, solver, solution_uid) - .await - { - Ok(Some(transaction)) => return Ok(transaction), - Ok(None) => {} - Err(err) => { - tracing::warn!( - ?err, - ?auction_id, - ?solver, - "failed to find settlement transaction" - ); - } - } - if block.number >= submission_deadline_latest_block { - break; - } - } - Err(SettleError::Timeout) - } } /// Picks a `/solve` deadline that ends shortly before a block gets @@ -1013,14 +849,6 @@ enum SolveError { SolverDenyListed, } -#[derive(Debug, thiserror::Error)] -enum SettleError { - #[error(transparent)] - Other(anyhow::Error), - #[error("settlement transaction await reached deadline")] - Timeout, -} - #[derive(prometheus_metric_storage::MetricStorage)] #[metric(subsystem = "runloop")] struct Metrics { diff --git a/crates/autopilot/src/run_loop/settle_call_coordinator.rs b/crates/autopilot/src/run_loop/settle_call_coordinator.rs new file mode 100644 index 0000000000..5015522c3c --- /dev/null +++ b/crates/autopilot/src/run_loop/settle_call_coordinator.rs @@ -0,0 +1,217 @@ +//! Coordinates a `/settle` call against a driver: enforces the block-based +//! submission deadline, races the driver response against the on-chain +//! observation of the resulting transaction, and journals both edges of the +//! call to the settlements-execution table. +//! +//! Extracted so the regular auction loop and the fast-path handler can share +//! the exact same submission machinery. + +use { + crate::{ + domain::settlement::{ExecutionEnded, ExecutionStarted}, + infra::{self, solvers::dto::settle}, + maintenance::{MaintenanceSync, SyncTarget}, + }, + eth_domain_types::{self as eth, TxId}, + futures::FutureExt, + std::time::Duration, + tracing::instrument, +}; + +#[derive(Debug, thiserror::Error)] +pub enum SettleError { + #[error(transparent)] + Other(anyhow::Error), + #[error("settlement transaction await reached deadline")] + Timeout, +} + +pub struct SettleCallCoordinator { + eth: infra::Ethereum, + persistence: infra::Persistence, + maintenance: MaintenanceSync, + /// How long we wait for the driver to signal the settlement completed + /// before considering the call timed out. + max_settlement_transaction_wait: Duration, +} + +impl SettleCallCoordinator { + pub fn new( + eth: infra::Ethereum, + persistence: infra::Persistence, + maintenance: MaintenanceSync, + max_settlement_transaction_wait: Duration, + ) -> Self { + Self { + eth, + persistence, + maintenance, + max_settlement_transaction_wait, + } + } + + /// Sends a `/settle` request to `driver` and returns the tx id of the + /// resulting settlement once it is mined (or an error if the submission + /// deadline was crossed first). + pub async fn settle( + &self, + driver: &infra::Driver, + solver: eth::Address, + solution_uid: usize, + request: settle::Request, + ) -> Result { + let auction_id = request.auction_id; + let deadline = request.submission_deadline_latest_block; + + let settle = async move { + let current_block = self.eth.current_block().borrow().number; + anyhow::ensure!( + current_block < request.submission_deadline_latest_block, + "submission deadline was missed" + ); + + self.store_execution_started( + request.auction_id, + solver, + solution_uid, + current_block, + request.submission_deadline_latest_block, + ); + driver + .settle(&request, self.max_settlement_transaction_wait) + .await + } + .boxed(); + + let wait_for_settlement_transaction = self + .wait_for_settlement_transaction(auction_id, solver, deadline, solution_uid) + .boxed(); + + // Wait for either the settlement transaction to be mined or the driver + // returned a result. + let result = match futures::future::select(wait_for_settlement_transaction, settle).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::Other(err)), + } + } + }; + + self.store_execution_ended(solver, auction_id, solution_uid, &result); + + result + } + + /// Stores settlement execution started event in the DB in a background + /// task to not block the caller. + fn store_execution_started( + &self, + auction_id: i64, + solver: eth::Address, + solution_uid: usize, + start_block: u64, + deadline_block: u64, + ) { + let persistence = self.persistence.clone(); + tokio::spawn(async move { + let execution_started = ExecutionStarted { + auction_id, + solver, + solution_uid, + start_timestamp: chrono::Utc::now(), + start_block, + deadline_block, + }; + + if let Err(err) = persistence + .store_settlement_execution_started(execution_started) + .await + { + tracing::error!(?err, "failed to store settlement execution event"); + } + }); + } + + /// Stores settlement execution ended event in the DB in a background task + /// to not block the caller. + fn store_execution_ended( + &self, + solver: eth::Address, + auction_id: i64, + solution_uid: usize, + result: &Result, + ) { + let end_timestamp = chrono::Utc::now(); + let current_block = self.eth.current_block().borrow().number; + let persistence = self.persistence.clone(); + let outcome = match result { + Ok(_) => "success".to_string(), + Err(SettleError::Timeout) => "timeout".to_string(), + Err(SettleError::Other(err)) => format!("driver failed: {err}"), + }; + + tokio::spawn(async move { + let execution_ended = ExecutionEnded { + auction_id, + solver, + solution_uid, + end_timestamp, + end_block: current_block, + outcome, + }; + if let Err(err) = persistence + .store_settlement_execution_ended(execution_ended) + .await + { + tracing::error!(?err, "failed to update settlement execution event"); + } + }); + } + + /// Tries to find a `settle` contract call originating from `solver` for + /// this auction on chain. Returns `Timeout` once the submission deadline + /// block passes without a match. + #[instrument(skip_all)] + async fn wait_for_settlement_transaction( + &self, + auction_id: i64, + solver: eth::Address, + submission_deadline_latest_block: u64, + solution_uid: usize, + ) -> Result { + let current = self.eth.current_block().borrow().number; + tracing::debug!(%current, deadline=%submission_deadline_latest_block, %auction_id, "waiting for tag"); + loop { + let block = ethrpc::block_stream::next_block(self.eth.current_block()).await; + // Run maintenance to ensure the system processed the last available + // block so it's possible to find the tx in the DB in + // the next line. + self.maintenance + .wait_until_block_processed(SyncTarget::FullyProcessed(block.number)) + .await; + + match self + .persistence + .find_settlement_transaction(auction_id, solver, solution_uid) + .await + { + Ok(Some(transaction)) => return Ok(transaction), + Ok(None) => {} + Err(err) => { + tracing::warn!( + ?err, + ?auction_id, + ?solver, + "failed to find settlement transaction" + ); + } + } + if block.number >= submission_deadline_latest_block { + break; + } + } + Err(SettleError::Timeout) + } +}