Only stream 1 /solve request body in driver (#4160)

MartinquaXD · squadgazzz · web-flow · commit 8f3b76cc6ec4 · 2026-02-17T11:05:02.000Z
# Description Because the driver serves multiple solvers it receives a bunch of duplicated `/solve` requests. There is already logic to deduplicate the pre-processing but we there is still one part left that's done unnecessarily often: streaming the HTTP body. Streaming the http body currently takes up to 700ms which is surprisingly slow considering that the HTTP request goes from one k8s pod to another and not via the public internet. I suspect the problem is that we are actually streaming ~10MB `/solve` requests 23 times in parallel (numbers from mainnet). #4159 introduced a new header (`X-Auction-Id`) that can be used to detect which auction a request is related to without having to stream the entire body. With this change everything but prioritizing (i.e. sorting and allocating balances for orders) and the serialization of the driver's `/solve` request will be de-duplicated. That means adding more solvers to the driver will be less costly. If we consider enforcing the same prioritization logic for ALL solvers that could also be de-duplicated leading to more or less 0 overhead for adding more solvers to the same driver. # Changes - inspect `X-Auction-Id` header to figure out whether we have to process the request or just await an existing pre-processing task Note that this change must be released AFTER https://github.com/cowprotocol/services/pull/4159`. The reason is that k8s first rolls out `driver` pods so there would be a period where the old `autopilot` is still sending requests without the `X-Auction-Id` header. ## How to test e2e tests --------- Co-authored-by: ilya <ilya@cow.fi>
diff --git a/crates/driver/src/domain/competition/mod.rs b/crates/driver/src/domain/competition/mod.rs
@@ -22,8 +22,9 @@ use {
         util::math,
     },
     alloy::primitives::Bytes,
+    axum::body::Body,
     futures::{StreamExt, future::Either, stream::FuturesUnordered},
-    hyper::body::Bytes as RequestBytes,
+    hyper::Request,
     itertools::Itertools,
     std::{
         cmp::Reverse,
@@ -114,14 +115,14 @@ impl Competition {
     }
 
     /// Solve an auction as part of this competition.
-    pub async fn solve(&self, auction: RequestBytes) -> Result<Option<Solved>, Error> {
+    pub async fn solve(&self, request: Request<Body>) -> Result<Option<Solved>, Error> {
         let start = Instant::now();
         let timer = ::observe::metrics::metrics()
             .on_auction_overhead_start("driver", "pre_processing_total");
 
         let tasks = self
             .fetcher
-            .start_or_get_tasks_for_auction(auction)
+            .start_or_get_tasks_for_auction(request)
             .await
             .map_err(|err| {
                 tracing::error!(?err, "pre-processing auction failed");
diff --git a/crates/driver/src/domain/competition/pre_processing.rs b/crates/driver/src/domain/competition/pre_processing.rs
@@ -11,9 +11,10 @@ use {
     },
     alloy::primitives::{Bytes, FixedBytes},
     anyhow::{Context, Result},
+    axum::body::Body,
     chrono::Utc,
     futures::{FutureExt, StreamExt, future::BoxFuture, stream::FuturesUnordered},
-    hyper::body::Bytes as RequestBytes,
+    hyper::{Request, body::Bytes as RequestBytes},
     itertools::Itertools,
     model::{
         interaction::InteractionData,
@@ -25,9 +26,14 @@ use {
         price_estimation::trade_verifier::balance_overrides::BalanceOverrideRequest,
         signature_validator::SignatureValidating,
     },
-    std::{collections::HashMap, future::Future, sync::Arc, time::Duration},
+    std::{
+        collections::HashMap,
+        future::Future,
+        sync::Arc,
+        time::{Duration, Instant},
+    },
     tokio::sync::Mutex,
-    tracing::Instrument,
+    tracing::{Instrument, instrument},
 };
 
 type Shared<T> = futures::future::Shared<BoxFuture<'static, T>>;
@@ -74,7 +80,7 @@ impl std::fmt::Debug for Utilities {
 #[derive(Debug)]
 struct ControlBlock {
     /// Auction for which the data aggregation task was spawned.
-    solve_request: RequestBytes,
+    auction_id: i64,
     /// Data aggregation task.
     tasks: DataFetchingTasks,
 }
@@ -91,26 +97,34 @@ impl DataAggregator {
     /// only once for all connected solvers to share.
     pub async fn start_or_get_tasks_for_auction(
         &self,
-        request: RequestBytes,
+        request: Request<Body>,
     ) -> Result<DataFetchingTasks> {
         let mut lock = self.control.lock().await;
-        let current_auction = &lock.solve_request;
-
-        // The autopilot ensures that all drivers receive identical
-        // requests per auction. That means we can use the significantly
-        // cheaper string comparison instead of parsing the JSON to compare
-        // the auction ids.
-        if request == current_auction {
-            let id = lock.tasks.auction.clone().await.id;
-            init_auction_id_in_span(id.map(|i| i.0));
+        let current_auction = &lock.auction_id;
+
+        // Figure out for which auction this `/solve` request was issued
+        // by looking at the `X-Auction-Id` header.
+        let request_auction_id: i64 = request
+            .headers()
+            .get("X-Auction-Id")
+            .context("request has no X-Auction-Id header")?
+            .to_str()
+            .context("X-Auction-Id header is not ASCII")?
+            .parse()
+            .context("could not parse X-Auction-Id header as i64")?;
+
+        // Some other driver is already doing the pre-processing for this
+        // auction. Stop processing here and just await the existing task.
+        if request_auction_id == *current_auction {
+            init_auction_id_in_span(Some(request_auction_id));
             tracing::debug!("await running data aggregation task");
             return Ok(lock.tasks.clone());
         }
 
-        let tasks = self.assemble_tasks(request.clone()).await?;
+        let tasks = self.assemble_tasks(request).await?;
 
         tracing::debug!("started new data aggregation task");
-        lock.solve_request = request;
+        lock.auction_id = request_auction_id;
         lock.tasks = tasks.clone();
 
         Ok(tasks)
@@ -153,7 +167,7 @@ impl DataAggregator {
                 cow_amm_cache,
             }),
             control: Mutex::new(ControlBlock {
-                solve_request: Default::default(),
+                auction_id: Default::default(),
                 tasks: DataFetchingTasks {
                     auction: futures::future::pending().boxed().shared(),
                     balances: futures::future::pending().boxed().shared(),
@@ -165,7 +179,7 @@ impl DataAggregator {
         }
     }
 
-    async fn assemble_tasks(&self, request: RequestBytes) -> Result<DataFetchingTasks> {
+    async fn assemble_tasks(&self, request: Request<Body>) -> Result<DataFetchingTasks> {
         let auction = self.utilities.parse_request(request).await?;
 
         let balances =
@@ -212,7 +226,9 @@ impl Utilities {
     /// Parses the JSON body of the `/solve` request during the unified
     /// auction pre-processing since eagerly deserializing these requests
     /// is surprisingly costly because their are so big.
-    async fn parse_request(&self, solve_request: RequestBytes) -> Result<Arc<Auction>> {
+    async fn parse_request(&self, solve_request: Request<Body>) -> Result<Arc<Auction>> {
+        let solve_request = collect_request_body(solve_request).await?;
+
         let auction_dto: SolveRequest = {
             let _timer = metrics::get().processing_stage_timer("parse_dto");
             let _timer2 =
@@ -539,3 +555,19 @@ fn init_auction_id_in_span(id: Option<i64>) {
     debug_assert!(current_span.has_field("auction_id"));
     current_span.record("auction_id", id);
 }
+
+#[instrument(skip_all)]
+async fn collect_request_body(request: Request<Body>) -> Result<RequestBytes> {
+    tracing::trace!("start streaming request body");
+    let _timer =
+        observe::metrics::metrics().on_auction_overhead_start("driver", "stream_http_body");
+    let start = Instant::now();
+
+    let body_bytes = hyper::body::to_bytes(request.into_body())
+        .await
+        .context("failed to stream request body")?;
+
+    let duration = start.elapsed();
+    tracing::debug!(?duration, "finished streaming request body");
+    Ok(body_bytes)
+}
diff --git a/crates/driver/src/infra/api/routes/solve/mod.rs b/crates/driver/src/infra/api/routes/solve/mod.rs
@@ -2,16 +2,11 @@ pub mod dto;
 
 pub use dto::AuctionError;
 use {
-    crate::{
-        domain::competition,
-        infra::{
-            api::{Error, State},
-            observe,
-        },
+    crate::infra::{
+        api::{Error, State},
+        observe,
     },
     axum::{body::Body, http::Request},
-    hyper::body::Bytes,
-    std::time::{Duration, Instant},
     tracing::Instrument,
 };
 
@@ -29,9 +24,8 @@ async fn route(
     let solver = state.solver().name().as_str();
 
     let handle_request = async {
-        let body_bytes = collect_request_body(request, solver).await?;
         let competition = state.competition();
-        let result = competition.solve(body_bytes).await;
+        let result = competition.solve(request).await;
         // Solving takes some time, so there is a chance for the settlement queue to
         // have capacity again.
         competition.ensure_settle_queue_capacity()?;
@@ -50,46 +44,3 @@ async fn route(
         ))
         .await
 }
-
-async fn collect_request_body(
-    request: Request<Body>,
-    solver: &str,
-) -> Result<Bytes, competition::Error> {
-    tracing::trace!("start streaming request body");
-    let start = Instant::now();
-
-    let body_bytes = hyper::body::to_bytes(request.into_body())
-        .await
-        .map_err(|err| {
-            tracing::warn!(?err, "failed to stream request body");
-            competition::Error::MalformedRequest
-        })?;
-
-    let duration = start.elapsed();
-    Metrics::measure_solve_transfer_time(solver, duration);
-    tracing::trace!(?duration, "finished streaming request body");
-    Ok(body_bytes)
-}
-
-#[derive(prometheus_metric_storage::MetricStorage)]
-struct Metrics {
-    /// Time spent by the driver reading the full solve request body into
-    /// memory.
-    #[metric(labels("solver"))]
-    #[metric(buckets(0.0001, 0.0005, 0.002, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 0.75, 1, 1.5))]
-    solve_request_body_read_duration_seconds: prometheus::HistogramVec,
-}
-
-impl Metrics {
-    fn get() -> &'static Metrics {
-        Metrics::instance(::observe::metrics::get_storage_registry())
-            .expect("unexpected error getting metrics instance")
-    }
-
-    fn measure_solve_transfer_time(solver: &str, time: Duration) {
-        Self::get()
-            .solve_request_body_read_duration_seconds
-            .with_label_values(&[solver])
-            .observe(time.as_secs_f64());
-    }
-}
diff --git a/crates/driver/src/tests/setup/mod.rs b/crates/driver/src/tests/setup/mod.rs
@@ -1059,6 +1059,7 @@ impl Test {
         let res = self
             .client
             .post(format!("http://{}/{}/solve", self.driver.addr, solver))
+            .header("X-Auction-Id", self.auction_id)
             .json(&driver::solve_req(self))
             .send()
             .await
