tcp tile for bid submissions

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 default-members = ["crates/relay"]
-members = ["crates/common", "crates/relay", "crates/simulator", "crates/types"]
+members = ["crates/common", "crates/relay", "crates/simulator", "crates/tcp-types", "crates/types"]
 resolver = "2"
 
 [workspace.package]
@@ -24,6 +24,7 @@ askama = "0.12.0"
 async-trait = "0.1"
 axum = { version = "0.8", features = ["matched-path", "ws"] }
 backtrace = "0.3.69"
+bitflags = "2.10.0"
 blst = { version = "0.3.16", features = ["no-threads"] }
 bytes = "1.10"
 chrono = { features = ["serde"], version = "0.4.38" }
@@ -40,10 +41,16 @@ ethereum_ssz_derive = "0.9"
 ethers = "2.0.14"
 eyre = "0.6.12"
 flate2 = "1.0"
-flux = { git = "https://github.com/gattaca-com/flux" }
+
+flux = { git = "https://github.com/gattaca-com/flux", rev="1fbb6568ee3635e3763e7dec52625b01a673837f"}
+flux-network = { package = "flux-network", git = "https://github.com/gattaca-com/flux", rev = "1fbb6568ee3635e3763e7dec52625b01a673837f"}
+flux-type-hash = { package = "type-hash", git = "https://github.com/gattaca-com/flux", rev = "1fbb6568ee3635e3763e7dec52625b01a673837f"}
+flux-type-hash-derive = { package = "type-hash-derive", git = "https://github.com/gattaca-com/flux", rev = "1fbb6568ee3635e3763e7dec52625b01a673837f"}
+
 futures = "0.3"
 futures-util = { version = "0.3", features = ["compat"] }
 helix-common = { path = "crates/common" }
+helix-tcp-types = { path = "./crates/tcp-types" }
 helix-types = { path = "./crates/types" }
 http = "1.0"
 http-body = "1.0"

crates/common/src/config.rs

@@ -59,6 +59,10 @@ pub struct RelayConfig {
     pub gossip_payload_on_header: bool,
     #[serde(default = "default_u16::<4040>")]
     pub api_port: u16,
+    #[serde(default = "default_u16::<4041>")]
+    pub tcp_port: u16,
+    #[serde(default = "default_usize::<512>")]
+    pub tcp_max_connections: usize,
 }
 
 impl RelayConfig {
@@ -89,9 +93,12 @@ impl RelayConfig {
                 tokio: vec![],
                 sub_workers: vec![],
                 reg_workers: vec![],
+                tcp_bid_submissions_tile: 2,
             },
             gossip_payload_on_header: false,
             api_port: 4040,
+            tcp_port: 4041,
+            tcp_max_connections: 512,
         }
     }
 }
@@ -134,6 +141,7 @@ pub struct CoresConfig {
     pub sub_workers: Vec<usize>,
     /// Registrations
     pub reg_workers: Vec<usize>,
+    pub tcp_bid_submissions_tile: usize,
 }
 
 impl Default for WebsiteConfig {

crates/common/src/local_cache.rs

@@ -10,7 +10,6 @@ use axum::{
 };
 use dashmap::{DashMap, DashSet};
 use helix_types::{BlsPublicKeyBytes, CryptoError, MergedBlock};
-use http::HeaderValue;
 use parking_lot::RwLock;
 use tracing::error;
 
@@ -89,7 +88,7 @@ pub struct LocalCache {
     pub inclusion_list: Arc<RwLock<Option<InclusionListWithKey>>>,
     builder_info_cache: Arc<DashMap<BlsPublicKeyBytes, BuilderInfo>>,
     /// Api key -> builder pubkey
-    api_key_cache: Arc<DashMap<HeaderValue, Vec<BlsPublicKeyBytes>>>,
+    pub api_key_cache: Arc<DashMap<String, Vec<BlsPublicKeyBytes>>>,
     trusted_proposers: Arc<DashMap<BlsPublicKeyBytes, ProposerInfo>>,
     primev_proposers: Arc<DashSet<BlsPublicKeyBytes>>,
     kill_switch: Arc<AtomicBool>,
@@ -119,28 +118,18 @@ impl LocalCache {
             merged_blocks,
         }
     }
-
-    pub fn new_test() -> Self {
-        Self::new()
-    }
-}
-
-impl Default for LocalCache {
-    fn default() -> Self {
-        Self::new()
-    }
 }
 
 impl LocalCache {
     pub fn get_builder_info(&self, builder_pub_key: &BlsPublicKeyBytes) -> Option<BuilderInfo> {
         Some(self.builder_info_cache.get(builder_pub_key)?.clone())
     }
 
-    pub fn contains_api_key(&self, api_key: &HeaderValue) -> bool {
+    pub fn contains_api_key(&self, api_key: &str) -> bool {
         self.api_key_cache.contains_key(api_key)
     }
 
-    pub fn validate_api_key(&self, api_key: &HeaderValue, pubkey: &BlsPublicKeyBytes) -> bool {
+    pub fn validate_api_key(&self, api_key: &str, pubkey: &BlsPublicKeyBytes) -> bool {
         self.api_key_cache.get(api_key).is_some_and(|p| p.value().contains(pubkey))
     }
 
@@ -167,10 +156,7 @@ impl LocalCache {
 
         for builder_info in builder_infos {
             if let Some(api_key) = builder_info.builder_info.api_key.as_ref() {
-                self.api_key_cache
-                    .entry(HeaderValue::from_str(api_key).unwrap())
-                    .or_default()
-                    .push(builder_info.pub_key);
+                self.api_key_cache.entry(api_key.clone()).or_default().push(builder_info.pub_key);
             }
 
             self.builder_info_cache.insert(builder_info.pub_key, builder_info.builder_info.clone());

crates/relay/Cargo.toml

@@ -26,8 +26,10 @@ ethers.workspace = true
 eyre.workspace = true
 flate2.workspace = true
 flux.workspace = true
+flux-network.workspace = true
 futures.workspace = true
 helix-common.workspace = true
+helix-tcp-types.workspace = true
 helix-types.workspace = true
 http.workspace = true
 http-body.workspace = true

crates/relay/src/api/admin_service.rs

@@ -47,7 +47,7 @@ mod test {
     #[tokio::test]
     #[serial]
     async fn test_admin_service() {
-        let auctioneer = Arc::new(LocalCache::new_test());
+        let auctioneer = Arc::new(LocalCache::new());
 
         let admin_token = "test_token".into();
         tokio::spawn(run_admin_service(auctioneer.clone(), admin_token));
@@ -76,7 +76,7 @@ mod test {
     #[tokio::test]
     #[serial]
     async fn test_admin_service_unauthorized() {
-        let auctioneer = Arc::new(LocalCache::new_test());
+        let auctioneer = Arc::new(LocalCache::new());
 
         let admin_token = "test_token".into();
         tokio::spawn(run_admin_service(auctioneer.clone(), admin_token));

crates/relay/src/api/builder/error.rs

@@ -1,6 +1,6 @@
 use axum::response::{IntoResponse, Response};
 use helix_common::{local_cache::AuctioneerError, simulator::BlockSimError};
-use helix_types::{BlockValidationError, HydrationError, SigError};
+use helix_types::{BlockValidationError, BlsPublicKeyBytes, HydrationError, SigError};
 use http::StatusCode;
 
 use crate::{auctioneer::OrderValidationError, database::error::DatabaseError};
@@ -60,6 +60,11 @@ pub enum BuilderApiError {
 
     #[error("internal error")]
     InternalError,
+
+    #[error(
+        "submission pubkey doesn't match registration message pubkey; expected: {0}, receieved: {1}"
+    )]
+    InvalidBuilderPubkey(BlsPublicKeyBytes, BlsPublicKeyBytes),
 }
 
 impl IntoResponse for BuilderApiError {
@@ -75,6 +80,7 @@ impl IntoResponse for BuilderApiError {
             BuilderApiError::SigError(_) |
             BuilderApiError::SimOnNextSlot |
             BuilderApiError::MergeableOrdersNotFound(_) |
+            BuilderApiError::InvalidBuilderPubkey(_, _) |
             BuilderApiError::DeliveringPayload { .. } => StatusCode::BAD_REQUEST,
 
             BuilderApiError::InvalidApiKey |

crates/relay/src/api/builder/submit_block.rs

@@ -8,14 +8,17 @@ use http::HeaderMap;
 use tracing::{error, trace};
 
 use super::api::BuilderApi;
-use crate::api::{Api, builder::error::BuilderApiError};
+use crate::{
+    api::{Api, builder::error::BuilderApiError},
+    auctioneer::{BlockSubResultSender, headers_map_to_bid_submission_header},
+};
 
 impl<A: Api> BuilderApi<A> {
     /// Implements this API: <https://flashbots.github.io/relay-specs/#/Builder/submitBlock>
     #[tracing::instrument(skip_all, err(level = tracing::Level::TRACE),
         fields(
-        id =% extract_request_id(&headers),
-        slot = tracing::field::Empty, // submission slot
+        id = tracing::field::Empty,
+        slot = tracing::field::Empty,
         builder_pubkey = tracing::field::Empty,
         builder_id = tracing::field::Empty,
         block_hash = tracing::field::Empty,
@@ -26,18 +29,39 @@ impl<A: Api> BuilderApi<A> {
         headers: HeaderMap,
         body: bytes::Bytes,
     ) -> Result<(), BuilderApiError> {
+        let request_id = extract_request_id(&headers);
+
+        tracing::Span::current().record("id", tracing::field::display(request_id));
+
         trace!("start handler");
 
         let mut trace = SubmissionTrace::init_from_timings(timings);
         trace.metadata = api.api_provider.get_metadata(&headers);
 
-        let Ok(rx) = api.auctioneer_handle.block_submission(headers, body, trace) else {
+        let (header, submission_ref, encoding, compression, api_key) =
+            headers_map_to_bid_submission_header(headers);
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        if api
+            .auctioneer_handle
+            .block_submission(
+                submission_ref,
+                header,
+                encoding,
+                compression,
+                api_key,
+                body,
+                trace,
+                BlockSubResultSender::OneShot(tx),
+                None,
+            )
+            .is_err()
+        {
             error!("failed sending request to worker");
             return Err(BuilderApiError::InternalError);
-        };
+        }
 
         let res = match rx.await {
-            Ok(res) => res,
+            Ok((_, res)) => res,
             Err(_) => Err(BuilderApiError::RequestTimeout),
         };
 

crates/relay/src/api/builder/top_bid.rs

@@ -22,7 +22,10 @@ impl<A: Api> BuilderApi<A> {
         headers: HeaderMap,
         ws: WebSocketUpgrade,
     ) -> Result<impl IntoResponse, BuilderApiError> {
-        let Some(api_key) = headers.get(HEADER_API_KEY).or_else(|| headers.get(HEADER_API_TOKEN))
+        let Some(api_key) = headers
+            .get(HEADER_API_KEY)
+            .or_else(|| headers.get(HEADER_API_TOKEN))
+            .and_then(|key| key.to_str().ok())
         else {
             return Err(BuilderApiError::InvalidApiKey);
         };

crates/relay/src/auctioneer/context.rs

@@ -214,7 +214,7 @@ impl<B: BidAdjustor> Context<B> {
         if let Some(res_tx) = result.res_tx {
             // submission was initially valid but by the time sim finished the slot already
             // progressed
-            let _ = res_tx.send(Err(BuilderApiError::SimOnNextSlot));
+            res_tx.try_send((result.submission_ref, Err(BuilderApiError::SimOnNextSlot)));
         }
     }