Skip to content

(fast5) extract /settle handling into separate component #4860

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MartinquaXD wants to merge 1 commit into from
+241 −196
Open

(fast5) extract /settle handling into separate component #4860

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
220 changes: 24 additions & 196 deletions crates/autopilot/src/run_loop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
pub mod settle_call_coordinator;

use {
self::settle_call_coordinator::{SettleCallCoordinator, SettleError},
crate::{
domain::{
self,
Expand All @@ -9,7 +12,6 @@ use {
Unscored,
winner_selection::{self, Ranking},
},
settlement::{ExecutionEnded, ExecutionStarted},
},
infra::{
self,
Expand All @@ -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,
Expand Down Expand Up @@ -166,6 +168,9 @@ pub struct RunLoop {

/// Drivers that do NOT support delta auctions
drivers: Vec<Arc<infra::Driver>>,
/// Sends `/settle` calls to drivers and waits for the resulting
/// transaction to be mined.
settle_coordinator: Arc<SettleCallCoordinator>,
}

impl RunLoop {
Expand All @@ -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,
Expand All @@ -200,6 +212,7 @@ impl RunLoop {
winner_selection: winner_selection::Arbitrator::new(max_winners, weth),
wake_notify: wake_runloop,
drivers,
settle_coordinator,
}
}

Expand Down Expand Up @@ -487,15 +500,15 @@ impl RunLoop {
tracing::info!(driver = %driver_.name, solution = %solution_id, "settling");
let submission_start = Instant::now();

let request = settle::Request {

@fleupold fleupold Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nanonit, would move request construction into the coordinator to avoid the settle dependency in runloop altogether.

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) => {
Expand Down Expand Up @@ -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<TxId, SettleError> {
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<TxId, SettleError>,
) {
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<eth::TxId, SettleError> {
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
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading