autopilot: send /solve requests with ref-counted body

Cargo.toml

@@ -60,6 +60,7 @@ autopilot = { path = "crates/autopilot" }
 aws-config = "1.5.1"
 aws-sdk-s3 = { version = "1.34.0", default-features = false }
 bytes-hex = { path = "crates/bytes-hex" }
+bytes = "1.11.1"
 chain = { path = "crates/chain" }
 console-subscriber = "0.3.0"
 const_format = "0.2.32"

crates/autopilot/Cargo.toml

@@ -19,6 +19,7 @@ alloy = { workspace = true, features = ["rand", "provider-debug-api", "provider-
 app-data = { workspace = true }
 axum = { workspace = true }
 bytes-hex = { workspace = true } # may get marked as unused but it's used with serde
+bytes = { workspace = true }
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 bigdecimal = { workspace = true }
@@ -47,7 +48,7 @@ order-validation = { workspace = true }
 prometheus = { workspace = true }
 prometheus-metric-storage = { workspace = true }
 rand = { workspace = true }
-reqwest = { workspace = true, features = ["gzip", "json"] }
+reqwest = { workspace = true, features = ["gzip", "json", "stream"] }
 rust_decimal = { workspace = true }
 s3 = { workspace = true }
 serde = { workspace = true }

crates/autopilot/src/infra/solvers/dto/solve.rs

@@ -2,17 +2,22 @@ use {
     crate::{
         boundary,
         domain::{self, eth},
-        infra::persistence::dto::{self, order::Order},
+        infra::{
+            persistence::dto::{self, order::Order},
+            solvers::InjectIntoHttpRequest,
+        },
     },
     alloy::primitives::{Address, U256},
+    bytes::Bytes,
     chrono::{DateTime, Utc},
     itertools::Itertools,
     number::serialization::HexOrDecimalU256,
+    reqwest::RequestBuilder,
     serde::{Deserialize, Serialize},
     serde_with::{DisplayFromStr, serde_as},
     std::{
+        borrow::Cow,
         collections::{HashMap, HashSet},
-        sync::Arc,
         time::Duration,
     },
 };
@@ -21,17 +26,14 @@ use {
 /// request. The purpose of this is to make it ergonomic
 /// to serialize a request once and reuse the resulting
 /// string in multiple HTTP requests.
-#[derive(Clone, Debug, Serialize, derive_more::Display)]
-pub struct Request(Arc<serde_json::value::RawValue>);
-
-impl Request {
-    pub fn as_str(&self) -> &str {
-        self.0.get()
-    }
+#[derive(Clone, Debug)]
+pub struct Request {
+    auction_id: i64,
+    body: bytes::Bytes,
 }
 
 impl Request {
-    pub fn new(
+    pub async fn new(
         auction: &domain::Auction,
         trusted_tokens: &HashSet<Address>,
         time_limit: Duration,
@@ -64,9 +66,38 @@ impl Request {
             deadline: Utc::now() + chrono::Duration::from_std(time_limit).unwrap(),
             surplus_capturing_jit_order_owners: auction.surplus_capturing_jit_order_owners.to_vec(),
         };
-        Self(Arc::from(serde_json::value::to_raw_value(&helper).expect(
-            "only fails with non-string keys which we do not have",
-        )))
+        let auction_id = auction.id;
+
+        let body = tokio::task::spawn_blocking(move || {
+            let serialized = serde_json::to_vec(&helper).expect("type should be JSON serializable");
+            Bytes::from(serialized)
+        })
+        .await
+        .expect("inner task should not panic as serialization should work for the given type");
+
+        Self { body, auction_id }
+    }
+}
+
+impl InjectIntoHttpRequest for Request {
+    fn inject(&self, request: RequestBuilder) -> RequestBuilder {
+        request
+            .body(self.body.clone())
+            // announce which auction this request is for in the
+            // headers to help the driver detect duplicated
+            // `/solve` requests before streaming the body
+            .header("X-Auction-Id", self.auction_id)
+            // manually set the content type header for JSON since
+            // we can't use `request.json(self)`
+            .header(
+                hyper::header::CONTENT_TYPE,
+                hyper::header::HeaderValue::from_static("application/json")
+            )
+    }
+
+    fn body_to_string(&self) -> Cow<'_, str> {
+        let string = str::from_utf8(self.body.as_ref()).unwrap();
+        Cow::Borrowed(string)
     }
 }
 

crates/autopilot/src/infra/solvers/mod.rs

@@ -4,8 +4,8 @@ use {
     alloy::signers::{Signer, aws::AwsSigner},
     anyhow::{Context, Result, anyhow},
     observe::tracing::tracing_headers,
-    reqwest::{Client, StatusCode},
-    std::time::Duration,
+    reqwest::{Client, RequestBuilder, StatusCode},
+    std::{borrow::Cow, time::Duration},
     thiserror::Error,
     tracing::instrument,
     url::Url,
@@ -122,28 +122,22 @@ impl Driver {
     async fn request_response<Response, Request>(
         &self,
         path: &str,
-        request: Request,
+        payload: Request,
     ) -> Result<Response>
     where
         Response: serde::de::DeserializeOwned,
-        Request: serde::Serialize + Send + Sync + 'static,
+        Request: InjectIntoHttpRequest,
     {
         let url = util::join(&self.url, path);
+
         tracing::trace!(
-            path=&url.path(),
-            body=%serde_json::to_string_pretty(&request).unwrap(),
+            path = &url.path(),
+            body = %payload.body_to_string(),
             "solver request",
         );
-        let mut request = {
-            let builder = self.client.post(url.clone()).headers(tracing_headers());
-            // If the payload is very big then serializing it will block the
-            // executor a long time (mostly relevant for solve requests).
-            // That's why we always do it on a thread specifically for
-            // running blocking tasks.
-            tokio::task::spawn_blocking(move || builder.json(&request))
-                .await
-                .context("failed to build request")?
-        };
+
+        let request = self.client.post(url.clone()).headers(tracing_headers());
+        let mut request = payload.inject(request);
 
         if let Some(request_id) = observe::distributed_tracing::request_id::from_current_span() {
             request = request.header("X-REQUEST-ID", request_id);
@@ -181,3 +175,22 @@ pub async fn response_body_with_size_limit(
     }
     Ok(bytes)
 }
+
+trait InjectIntoHttpRequest {
+    fn inject(&self, request: RequestBuilder) -> RequestBuilder;
+    fn body_to_string(&self) -> Cow<'_, str>;
+}
+
+impl<T> InjectIntoHttpRequest for T
+where
+    T: serde::ser::Serialize + Sized,
+{
+    fn inject(&self, request: RequestBuilder) -> RequestBuilder {
+        request.json(&self)
+    }
+
+    fn body_to_string(&self) -> Cow<'_, str> {
+        let serialized = serde_json::to_string(&self).expect("type should be JSON serializable");
+        Cow::Owned(serialized)
+    }
+}

crates/autopilot/src/run_loop.rs

@@ -561,7 +561,8 @@ impl RunLoop {
             auction,
             &self.trusted_tokens.all(),
             self.config.solve_deadline,
-        );
+        )
+        .await;
 
         let mut bids = futures::future::join_all(
             self.drivers

crates/autopilot/src/shadow.rs

@@ -179,7 +179,8 @@ impl RunLoop {
     /// Runs the solver competition, making all configured drivers participate.
     #[instrument(skip_all)]
     async fn competition(&self, auction: &domain::Auction) -> Vec<Bid<Unscored>> {
-        let request = solve::Request::new(auction, &self.trusted_tokens.all(), self.solve_deadline);
+        let request =
+            solve::Request::new(auction, &self.trusted_tokens.all(), self.solve_deadline).await;
 
         futures::future::join_all(
             self.drivers