Skip to content

Commit a1b49e8

Browse files
metalurgicalmetalurgical
authored
Merge branch 'main' into fix_chore_3516
2 parents e477d73 + bf40548 commit a1b49e8

30 files changed

Lines changed: 631 additions & 177 deletions

File tree

‎Cargo.lock‎

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎configs/local/orderbook.toml‎

Lines changed: 0 additions & 19 deletions
This file was deleted.

‎crates/driver/src/domain/competition/mod.rs‎

Lines changed: 107 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -46,12 +46,19 @@ pub mod sorting;
4646
use {
4747
crate::infra::notify::liquidity_sources::LiquiditySourceNotifying,
4848
eth_domain_types::BlockNo,
49+
ethrpc::block_stream::BlockInfo,
4950
};
5051
pub use {auction::Auction, order::Order, pre_processing::DataAggregator, solution::Solution};
5152

5253
type BalanceGroup = (order::Trader, eth::TokenAddress, order::SellTokenBalance);
5354
type Balances = HashMap<BalanceGroup, order::SellAmount>;
5455

56+
/// How many concurrent auction generations we size the settlement cache for.
57+
/// Each auction can propose up to `max_solutions_to_propose` settlements;
58+
/// keeping this many generations around lets a stale /settle for a previous
59+
/// auction still find its settlement.
60+
const MAX_CONCURRENT_AUCTIONS: usize = 5;
61+
5562
/// An ongoing competition. There is one competition going on per solver at any
5663
/// time. The competition stores settlements to solutions generated by the
5764
/// driver, and allows them to be executed onchain when requested later. The
@@ -304,7 +311,7 @@ impl Competition {
304311
}
305312

306313
/// Solve an auction as part of this competition.
307-
pub async fn solve(&self, request: Request<Body>) -> Result<Option<Solved>, Error> {
314+
pub async fn solve(&self, request: Request<Body>) -> Result<Vec<Solved>, Error> {
308315
let start = Instant::now();
309316
let timer = ::observe::metrics::metrics()
310317
.on_auction_overhead_start("driver", "pre_processing_total");
@@ -389,7 +396,7 @@ impl Competition {
389396

390397
if auction.orders.is_empty() {
391398
tracing::info!("no orders left after pre-processing; skipping solving");
392-
return Ok(None);
399+
return Ok(vec![]);
393400
}
394401

395402
let auction = &auction;
@@ -543,76 +550,116 @@ impl Competition {
543550
observe::score(settlement, score);
544551
}
545552

546-
// Pick the best-scoring settlement.
547-
let (mut score, settlement) = scores
553+
// Build scored settlements sorted best-first.
554+
let scored: Vec<(Solved, Settlement)> = scores
548555
.into_iter()
549-
.max_by_key(|(score, _)| score.to_owned())
556+
.sorted_by_key(|(score, _)| Reverse(*score))
550557
.map(|(score, settlement)| {
551-
(
552-
Solved {
553-
id: settlement.solution().clone(),
554-
score,
555-
trades: settlement.orders(),
556-
prices: settlement.prices(),
557-
gas: Some(settlement.gas.estimate),
558-
},
559-
settlement,
560-
)
558+
let solved = Solved {
559+
id: settlement.solution().clone(),
560+
score,
561+
trades: settlement.orders(),
562+
prices: settlement.prices(),
563+
gas: Some(settlement.gas.estimate),
564+
};
565+
(solved, settlement)
561566
})
562-
.unzip();
567+
.collect();
563568

564-
let Some(settlement) = settlement else {
565-
// Don't wait for the deadline because we can't produce a solution anyway.
566-
return Ok(score);
567-
};
568-
let solution_id = settlement.solution().get();
569+
if scored.is_empty() {
570+
return Ok(vec![]);
571+
}
569572

573+
let max_to_propose = self.solver.max_solutions_to_propose();
574+
let mut scored: Vec<(Solved, Settlement)> =
575+
scored.into_iter().take(max_to_propose).collect();
576+
577+
// Cache all settlements so they can be revealed/settled later. Keep
578+
// solutions from previous overlapping auctions around long enough
579+
// for their /settle to complete.
570580
{
571581
let mut lock = self.settlements.lock().unwrap();
572-
lock.push_front(settlement.clone());
573-
574-
/// Number of solutions that may be cached at most.
575-
const MAX_SOLUTION_STORAGE: usize = 5;
576-
lock.truncate(MAX_SOLUTION_STORAGE);
582+
for (_, settlement) in &scored {
583+
lock.push_front(settlement.clone());
584+
}
585+
lock.truncate(max_to_propose * MAX_CONCURRENT_AUCTIONS);
577586
}
578587

579-
// Re-simulate the solution on every new block until the deadline ends to make
580-
// sure we actually submit a working solution close to when the winner
581-
// gets picked by the procotol.
582588
if let Ok(remaining) = deadline.remaining() {
583-
let score_ref = &mut score;
584-
let has_haircut = settlement.has_haircut();
585-
let simulate_on_new_blocks = async move {
586-
let mut stream =
587-
ethrpc::block_stream::into_stream(self.eth.current_block().clone());
588-
while let Some(block) = stream.next().await {
589-
if let Err(simulator::Error::Revert(err)) =
590-
self.simulate_settlement(&settlement).await
591-
{
592-
observe::winner_voided(self.solver.name(), block, &err, has_haircut);
593-
*score_ref = None;
594-
self.settlements
595-
.lock()
596-
.unwrap()
597-
.retain(|s| s.solution().get() != solution_id);
598-
// Only notify solver if solution doesn't have haircut
599-
if !has_haircut {
600-
notify::simulation_failed(
601-
&self.solver,
602-
auction.id(),
603-
settlement.solution(),
604-
&simulator::Error::Revert(err),
605-
true,
606-
);
607-
}
608-
return;
609-
}
610-
}
611-
};
612-
let _ = tokio::time::timeout(remaining, simulate_on_new_blocks).await;
589+
let _ = tokio::time::timeout(
590+
remaining,
591+
self.resimulate_until_revert(&mut scored, auction),
592+
)
593+
.await;
594+
}
595+
596+
Ok(scored.into_iter().map(|(solved, _)| solved).collect())
597+
}
598+
599+
/// Re-simulate all proposed solutions on every new block and drop any
600+
/// that start reverting. Returns once every solution has reverted;
601+
/// otherwise runs forever and the caller must impose a deadline.
602+
/// Mutates `scored` and the cached settlements in place.
603+
async fn resimulate_until_revert(
604+
&self,
605+
scored: &mut Vec<(Solved, Settlement)>,
606+
auction: &Auction,
607+
) -> anyhow::Result<()> {
608+
let mut stream = ethrpc::block_stream::into_stream(self.eth.current_block().clone());
609+
while let Some(block) = stream.next().await {
610+
let voided_ids: HashSet<u64> =
611+
futures::future::join_all(scored.iter().map(|(solved, settlement)| {
612+
self.reverts_on_block(block, solved, settlement, auction)
613+
}))
614+
.await
615+
.into_iter()
616+
.flatten()
617+
.collect();
618+
619+
if voided_ids.is_empty() {
620+
continue;
621+
}
622+
623+
scored.retain(|(solved, _)| !voided_ids.contains(&solved.id.get()));
624+
let mut lock = self
625+
.settlements
626+
.lock()
627+
.map_err(|_| anyhow::anyhow!("settlements mutex poisoned"))?;
628+
lock.retain(|s| !voided_ids.contains(&s.solution().get()));
629+
630+
if scored.is_empty() {
631+
return Ok(());
632+
}
613633
}
634+
Ok(())
635+
}
614636

615-
Ok(score)
637+
/// Re-simulate a single solution and return its id if it started
638+
/// reverting on this block. Reports metrics and solver notifications as
639+
/// a side effect.
640+
async fn reverts_on_block(
641+
&self,
642+
block: BlockInfo,
643+
solved: &Solved,
644+
settlement: &Settlement,
645+
auction: &Auction,
646+
) -> Option<u64> {
647+
let err = match self.simulate_settlement(settlement).await {
648+
Err(simulator::Error::Revert(err)) => err,
649+
_ => return None,
650+
};
651+
let has_haircut = settlement.has_haircut();
652+
observe::winner_voided(self.solver.name(), block, &err, has_haircut);
653+
if !has_haircut {
654+
notify::simulation_failed(
655+
&self.solver,
656+
auction.id(),
657+
settlement.solution(),
658+
&simulator::Error::Revert(err),
659+
true,
660+
);
661+
}
662+
Some(solved.id.get())
616663
}
617664

618665
// Oders already need to be sorted from most relevant to least relevant so that

‎crates/driver/src/domain/competition/pre_processing.rs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -235,7 +235,11 @@ impl Utilities {
235235
observe::metrics::metrics().on_auction_overhead_start("driver", "parse_dto");
236236
// deserialization takes tens of milliseconds so run it on a blocking task
237237
tokio::task::spawn_blocking(move || {
238-
serde_json::from_slice(&solve_request).context("could not parse solve request")
238+
serde_json::from_slice(&solve_request)
239+
.inspect_err(|err| {
240+
tracing::warn!(?err, "failed to parse /solve request body");
241+
})
242+
.context("could not parse solve request")
239243
})
240244
.await
241245
.context("failed to await blocking task")??

0 commit comments

Comments
 (0)