diff --git a/crates/autopilot/src/infra/persistence/dto/auction.rs b/crates/autopilot/src/infra/persistence/dto/auction.rs index b3aa28fcbe..46dce7be84 100644 --- a/crates/autopilot/src/infra/persistence/dto/auction.rs +++ b/crates/autopilot/src/infra/persistence/dto/auction.rs @@ -1,5 +1,5 @@ use { - super::order::Order, + super::order::{Order, OrdersJson}, crate::domain::{self, auction::Price}, alloy::primitives::{Address, U256}, eth_domain_types as eth, @@ -11,16 +11,18 @@ use { /// Converts the auction into the shape that gets archived to the DB and S3. /// +/// The order list comes in already serialized because those bytes are shared +/// with the `/solve` request (see [`OrdersJson`]). +/// /// Takes the auction by reference so the caller can keep using it afterwards /// without a deep clone. -pub fn from_domain(auction: &domain::RawAuctionData) -> RawAuctionData { +pub fn from_domain( + auction: &domain::RawAuctionData, + orders: OrdersJson, +) -> RawAuctionData { RawAuctionData { block: auction.block, - orders: auction - .orders - .iter() - .map(super::order::from_domain) - .collect(), + orders, prices: auction .prices .iter() @@ -30,12 +32,19 @@ pub fn from_domain(auction: &domain::RawAuctionData) -> RawAuctionData { } } +/// The archived auction. Generic over the order list so the write path can +/// splice in the pre-rendered [`OrdersJson`] while the read path deserializes +/// into [`Order`]s. One definition, so the two shapes can't drift apart. +/// +/// `Deserialize` is only ever instantiated at `O = Vec`; the derive puts +/// the bound on the impl, not on the struct, so `RawAuctionData` +/// simply has no `Deserialize` impl. #[serde_as] #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] -pub struct RawAuctionData { +pub struct RawAuctionData> { pub block: u64, - pub orders: Vec, + pub orders: O, #[serde_as(as = "BTreeMap<_, HexOrDecimalU256>")] pub prices: BTreeMap, #[serde(default)] @@ -79,3 +88,38 @@ impl Auction { }) } } + +#[cfg(test)] +mod tests { + use super::*; + + /// The archived auction is written with pre-rendered orders and read back + /// into `Vec`, so the two shapes must stay in sync. + #[tokio::test] + async fn archived_auction_round_trips() { + let token = Address::from([1u8; 20]); + let auction = domain::RawAuctionData { + block: 42, + orders: vec![], + prices: [( + token.into(), + Price::try_new(U256::from(1000).into()).unwrap(), + )] + .into_iter() + .collect(), + surplus_capturing_jit_order_owners: vec![Address::from([2u8; 20])], + }; + + let written = from_domain(&auction, OrdersJson::new(&auction.orders).await); + let json = serde_json::to_string(&written).unwrap(); + + let read: RawAuctionData = serde_json::from_str(&json).unwrap(); + assert_eq!(read.block, 42); + assert!(read.orders.is_empty()); + assert_eq!(read.prices, written.prices); + assert_eq!( + read.surplus_capturing_jit_order_owners, + written.surplus_capturing_jit_order_owners + ); + } +} diff --git a/crates/autopilot/src/infra/persistence/dto/order.rs b/crates/autopilot/src/infra/persistence/dto/order.rs index ca5a9cd6bc..4cb93de1a1 100644 --- a/crates/autopilot/src/infra/persistence/dto/order.rs +++ b/crates/autopilot/src/infra/persistence/dto/order.rs @@ -8,8 +8,10 @@ use { configs::fee_factor::FeeFactor, eth_domain_types as eth, number::serialization::HexOrDecimalU256, - serde::{Deserialize, Serialize}, + serde::{Deserialize, Serialize, Serializer}, + serde_json::value::RawValue, serde_with::serde_as, + std::sync::Arc, }; #[serde_as] @@ -124,6 +126,44 @@ pub fn to_domain(order: Order) -> domain::Order { } } +/// The auction's order list, already rendered to JSON. Serializes verbatim so +/// the archived auction and the `/solve` request body can splice in the same +/// bytes instead of each converting and serializing the orders themselves. +/// Cheap to clone. +#[derive(Clone, Debug)] +pub struct OrdersJson(Arc); + +impl OrdersJson { + /// Converts the orders into their DTO shape on the calling thread (they are + /// borrowed, so this can't be moved into the background without a deep + /// clone) and renders them to JSON on the blocking pool. + pub async fn new(orders: &[domain::Order]) -> Self { + let orders: Vec = { + let _timer = observe::metrics::metrics() + .on_auction_overhead_start("autopilot", "convert_orders"); + orders.iter().map(from_domain).collect() + }; + + tokio::task::spawn_blocking(move || { + let _timer = observe::metrics::metrics() + .on_auction_overhead_start("autopilot", "serialize_orders"); + // `to_raw_value` keeps the serializer output as-is; unlike + // `RawValue::from_string` it doesn't re-scan the JSON to validate it. + let json = serde_json::value::to_raw_value(&orders) + .expect("orders should be JSON serializable"); + OrdersJson(Arc::from(json)) + }) + .await + .expect("order serialization should not panic") + } +} + +impl Serialize for OrdersJson { + fn serialize(&self, serializer: S) -> Result { + self.0.serialize(serializer) + } +} + impl From for domain::OrderUid { fn from(uid: boundary::OrderUid) -> Self { Self(uid.0) @@ -383,3 +423,35 @@ impl From for database::orders::OrderKind { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Serialize)] + #[serde(rename_all = "camelCase")] + struct Wrapper { + orders: OrdersJson, + block: u64, + } + + #[test] + fn orders_json_is_spliced_verbatim() { + let orders = serde_json::json!([{"uid": "0x01"}, {"uid": "0x02"}]); + let orders = OrdersJson(Arc::from(serde_json::value::to_raw_value(&orders).unwrap())); + + assert_eq!( + serde_json::to_string(&Wrapper { orders, block: 7 }).unwrap(), + r#"{"orders":[{"uid":"0x01"},{"uid":"0x02"}],"block":7}"# + ); + } + + #[tokio::test] + async fn empty_orders_json() { + let orders = OrdersJson::new(&[]).await; + + assert_eq!(orders.0.get(), "[]"); + // both consumers get the very same bytes instead of re-rendering them + assert!(Arc::ptr_eq(&orders.0, &orders.clone().0)); + } +} diff --git a/crates/autopilot/src/infra/persistence/mod.rs b/crates/autopilot/src/infra/persistence/mod.rs index 953e5c4b34..ff595ac38c 100644 --- a/crates/autopilot/src/infra/persistence/mod.rs +++ b/crates/autopilot/src/infra/persistence/mod.rs @@ -155,14 +155,20 @@ impl Persistence { /// /// Only the conversion into the archival shape is on the run loop's /// critical path; the serialization and both sinks happen in background - /// tasks. The auction is taken by reference so the conversion doesn't - /// need a deep clone. + /// tasks. The order list is not converted here at all: those bytes are + /// serialized once and shared with the `/solve` request. The auction is + /// taken by reference so the conversion doesn't need a deep clone. #[instrument(skip_all)] - pub fn archive_auction(&self, id: domain::auction::Id, auction: &domain::RawAuctionData) { + pub fn archive_auction( + &self, + id: domain::auction::Id, + auction: &domain::RawAuctionData, + orders: dto::order::OrdersJson, + ) { let auction_data = { let _timer = observe::metrics::metrics() .on_auction_overhead_start("autopilot", "convert_auction"); - dto::auction::from_domain(auction) + dto::auction::from_domain(auction, orders) }; let upload_to_s3 = !auction.orders.is_empty(); diff --git a/crates/autopilot/src/infra/solvers/dto/solve.rs b/crates/autopilot/src/infra/solvers/dto/solve.rs index 86e98e719e..e4d7218b1a 100644 --- a/crates/autopilot/src/infra/solvers/dto/solve.rs +++ b/crates/autopilot/src/infra/solvers/dto/solve.rs @@ -2,10 +2,7 @@ use { crate::{ boundary, domain, - infra::{ - persistence::dto::{self, order::Order}, - solvers::InjectIntoHttpRequest, - }, + infra::{persistence::dto::order::OrdersJson, solvers::InjectIntoHttpRequest}, }, alloy::primitives::{Address, U256}, brotli::enc::writer::CompressorWriter, @@ -42,6 +39,7 @@ pub struct Request { impl Request { pub async fn new( auction: &domain::Auction, + orders: OrdersJson, trusted_tokens: &HashSet
, deadline: chrono::DateTime, compress: bool, @@ -50,7 +48,7 @@ impl Request { observe::metrics::metrics().on_auction_overhead_start("autopilot", "serialize_request"); let helper = RequestHelper { id: auction.id, - orders: auction.orders.iter().map(dto::order::from_domain).collect(), + orders, tokens: auction .prices .iter() @@ -175,13 +173,13 @@ impl Response { } #[serde_as] -#[derive(Clone, Debug, Default, Serialize)] +#[derive(Clone, Debug, Serialize)] #[serde(rename_all = "camelCase")] struct RequestHelper { #[serde_as(as = "DisplayFromStr")] pub id: i64, pub tokens: Vec, - pub orders: Vec, + pub orders: OrdersJson, pub deadline: DateTime, pub surplus_capturing_jit_order_owners: Vec
, } diff --git a/crates/autopilot/src/run_loop.rs b/crates/autopilot/src/run_loop.rs index 4063673ff8..9fd5fbb412 100644 --- a/crates/autopilot/src/run_loop.rs +++ b/crates/autopilot/src/run_loop.rs @@ -14,6 +14,7 @@ use { }, infra::{ self, + persistence::dto::order::OrdersJson, solvers::dto::{settle, solve}, }, leader_lock_tracker::LeaderLockTracker, @@ -102,6 +103,13 @@ pub struct Probes { pub startup: Arc>, } +/// A freshly cut auction, carrying its order list already rendered to JSON so +/// the archived auction and every `/solve` request reuse the same bytes. +struct CutAuction { + auction: domain::Auction, + orders_json: OrdersJson, +} + pub struct RunLoop { config: Config, eth: infra::Ethereum, @@ -193,13 +201,13 @@ impl RunLoop { continue; } - if let Some(auction) = self_arc + if let Some(cut) = self_arc .next_auction(start_block, &mut last_auction, &mut last_block) .await { - let auction_id = auction.id; + let auction_id = cut.auction.id; self_arc - .single_run(auction) + .single_run(cut) .instrument(tracing::info_span!("auction", auction_id)) .await } @@ -282,26 +290,27 @@ impl RunLoop { start_block: BlockInfo, prev_auction: &mut Option, prev_block: &mut Option, - ) -> Option { + ) -> Option { // wait for appropriate time to start building the auction - let auction = self.cut_auction().await?; + let cut = self.cut_auction().await?; + let auction = &cut.auction; tracing::trace!(auction_id = ?auction.id, "auction cut"); // Only run the solvers if the auction or block has changed. let previous = prev_auction.replace(auction.clone()); - if previous.as_ref() == Some(&auction) + if previous.as_ref() == Some(auction) && prev_block.replace(start_block.hash) == Some(start_block.hash) { return None; } - observe::log_auction_delta(&previous, &auction, &start_block); + observe::log_auction_delta(&previous, auction, &start_block); self.probes.liveness.auction(); Metrics::auction_ready(start_block.observed_at); - Some(auction) + Some(cut) } - async fn cut_auction(&self) -> Option { + async fn cut_auction(&self) -> Option { let Some(auction) = self.solvable_orders_cache.current_auction().await else { tracing::debug!("no current auction"); return None; @@ -314,8 +323,11 @@ impl RunLoop { .ok()?; Metrics::auction(id); + let orders_json = OrdersJson::new(&auction.orders).await; + // always update the auction because the tests use this as a readiness probe - self.persistence.archive_auction(id, &auction); + self.persistence + .archive_auction(id, &auction, orders_json.clone()); if auction.orders.is_empty() { // Updating liveness probe to not report unhealthy due to this optimization @@ -323,18 +335,22 @@ impl RunLoop { tracing::debug!("skipping empty auction"); return None; } - Some(domain::Auction { - id, - block: auction.block, - orders: auction.orders, - prices: auction.prices, - surplus_capturing_jit_order_owners: auction.surplus_capturing_jit_order_owners, + Some(CutAuction { + auction: domain::Auction { + id, + block: auction.block, + orders: auction.orders, + prices: auction.prices, + surplus_capturing_jit_order_owners: auction.surplus_capturing_jit_order_owners, + }, + orders_json, }) } #[instrument(skip_all)] - async fn single_run(self: &Arc, auction: domain::Auction) { + async fn single_run(self: &Arc, cut: CutAuction) { let single_run_start = Instant::now(); + let auction = &cut.auction; tracing::info!(auction_id = ?auction.id, "solving"); // Mark all auction orders as `Ready` for competition @@ -343,13 +359,13 @@ impl RunLoop { tracing::trace!(auction_id = ?auction.id, "orders marked as ready"); // Collect valid solutions from all drivers - let solutions = self.fetch_solutions(&auction).await; + let solutions = self.fetch_solutions(&cut).await; observe::bids(&solutions); if solutions.is_empty() { return; } - let ranking = self.winner_selection.arbitrate(solutions, &auction); + let ranking = self.winner_selection.arbitrate(solutions, auction); // Count and record the number of winners let num_winners = ranking.winners().count(); @@ -364,7 +380,7 @@ impl RunLoop { // of storing all the competition/auction-related data to the DB. if let Err(err) = self .post_processing( - &auction, + auction, competition_simulation_block, &ranking, block_deadline, @@ -408,7 +424,7 @@ impl RunLoop { ); } tracing::trace!(auction_id = ?auction.id, "settlement execution started"); - observe::unsettled(&ranking, &auction); + observe::unsettled(&ranking, auction); } /// Starts settlement execution in a background task. The function is async @@ -588,11 +604,12 @@ impl RunLoop { /// Runs the solver competition, making all configured drivers participate. /// Returns all fair solutions sorted by their score (best to worst). #[instrument(skip_all)] - async fn fetch_solutions(&self, auction: &domain::Auction) -> Vec> { + async fn fetch_solutions(&self, cut: &CutAuction) -> Vec> { let deadline = self.pick_solve_deadline(); let request = solve::Request::new( - auction, + &cut.auction, + cut.orders_json.clone(), &self.trusted_tokens.all(), deadline, self.config.compress_solve_request, diff --git a/crates/autopilot/src/shadow.rs b/crates/autopilot/src/shadow.rs index aed38bdd27..1b6cd0769d 100644 --- a/crates/autopilot/src/shadow.rs +++ b/crates/autopilot/src/shadow.rs @@ -15,6 +15,7 @@ use { }, infra::{ self, + persistence::dto::order::OrdersJson, solvers::dto::{reveal, solve}, }, run::Liveness, @@ -184,6 +185,7 @@ impl RunLoop { async fn competition(&self, auction: &domain::Auction) -> Vec> { let request = solve::Request::new( auction, + OrdersJson::new(&auction.orders).await, &self.trusted_tokens.all(), Utc::now() + self.solve_deadline, self.compress_solve_request,