Merge remote-tracking branch 'eserilev/add-builder-ssz-flow' into add-ssz-to-pbs

Author: jclapis

crates/cli/src/docker_init.rs

@@ -154,9 +154,10 @@ pub async fn handle_docker_init(config_path: PathBuf, output_dir: PathBuf) -> Re
 
                     // depends_on
                     let mut module_dependencies = IndexMap::new();
-                    module_dependencies.insert("cb_signer".into(), DependsCondition {
-                        condition: "service_healthy".into(),
-                    });
+                    module_dependencies.insert(
+                        "cb_signer".into(),
+                        DependsCondition { condition: "service_healthy".into() },
+                    );
 
                     Service {
                         container_name: Some(module_cid.clone()),

crates/common/src/signer/store.rs

@@ -302,8 +302,8 @@ impl ProxyStore {
                                 let entry = entry?;
                                 let path = entry.path();
 
-                                if !path.is_file() ||
-                                    path.extension().is_none_or(|ext| ext != "json")
+                                if !path.is_file()
+                                    || path.extension().is_none_or(|ext| ext != "json")
                                 {
                                     continue;
                                 }
@@ -376,8 +376,8 @@ impl ProxyStore {
                                 let entry = entry?;
                                 let path = entry.path();
 
-                                if !path.is_file() ||
-                                    path.extension().is_none_or(|ext| ext != "json")
+                                if !path.is_file()
+                                    || path.extension().is_none_or(|ext| ext != "json")
                                 {
                                     continue;
                                 }

crates/common/src/types.rs

@@ -358,11 +358,14 @@ mod tests {
     fn test_load_custom() {
         let s = r#"chain = { genesis_time_secs = 1, slot_time_secs = 2, genesis_fork_version = "0x01000000" }"#;
         let decoded: MockConfig = toml::from_str(s).unwrap();
-        assert_eq!(decoded.chain, Chain::Custom {
-            genesis_time_secs: 1,
-            slot_time_secs: 2,
-            genesis_fork_version: [1, 0, 0, 0]
-        })
+        assert_eq!(
+            decoded.chain,
+            Chain::Custom {
+                genesis_time_secs: 1,
+                slot_time_secs: 2,
+                genesis_fork_version: [1, 0, 0, 0]
+            }
+        )
     }
 
     #[test]

crates/common/src/utils.rs

@@ -603,6 +603,192 @@ where
     }
 }
 
+/// Parse ACCEPT header, default to JSON if missing or mal-formatted
+pub fn get_accept_header(req_headers: &HeaderMap) -> Accept {
+    Accept::from_str(
+        req_headers.get(ACCEPT).and_then(|value| value.to_str().ok()).unwrap_or("application/json"),
+    )
+    .unwrap_or(Accept::Json)
+}
+
+/// Parse CONTENT TYPE header, default to JSON if missing or mal-formatted
+pub fn get_content_type_header(req_headers: &HeaderMap) -> ContentType {
+    ContentType::from_str(
+        req_headers
+            .get(CONTENT_TYPE)
+            .and_then(|value| value.to_str().ok())
+            .unwrap_or("application/json"),
+    )
+    .unwrap_or(ContentType::Json)
+}
+
+/// Parse CONSENSUS_VERSION header
+pub fn get_consensus_version_header(req_headers: &HeaderMap) -> Option<ForkName> {
+    ForkName::from_str(
+        req_headers
+            .get(CONSENSUS_VERSION_HEADER)
+            .and_then(|value| value.to_str().ok())
+            .unwrap_or(""),
+    )
+    .ok()
+}
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum ForkName {
+    Deneb,
+    Electra,
+}
+
+impl std::fmt::Display for ForkName {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ForkName::Deneb => write!(f, "deneb"),
+            ForkName::Electra => write!(f, "electra"),
+        }
+    }
+}
+
+impl FromStr for ForkName {
+    type Err = String;
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "deneb" => Ok(ForkName::Deneb),
+            "electra" => Ok(ForkName::Electra),
+            _ => Err(format!("Invalid fork name {}", value)),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum ContentType {
+    Json,
+    Ssz,
+}
+
+impl std::fmt::Display for ContentType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ContentType::Json => write!(f, "application/json"),
+            ContentType::Ssz => write!(f, "application/octet-stream"),
+        }
+    }
+}
+
+impl FromStr for ContentType {
+    type Err = String;
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "application/json" => Ok(ContentType::Json),
+            "application/octet-stream" => Ok(ContentType::Ssz),
+            _ => Ok(ContentType::Json),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum Accept {
+    Json,
+    Ssz,
+    Any,
+}
+
+impl fmt::Display for Accept {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Accept::Ssz => write!(f, "application/octet-stream"),
+            Accept::Json => write!(f, "application/json"),
+            Accept::Any => write!(f, "*/*"),
+        }
+    }
+}
+
+impl FromStr for Accept {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let media_type_list = MediaTypeList::new(s);
+
+        // [q-factor weighting]: https://datatracker.ietf.org/doc/html/rfc7231#section-5.3.2
+        // find the highest q-factor supported accept type
+        let mut highest_q = 0_u16;
+        let mut accept_type = None;
+
+        const APPLICATION: &str = names::APPLICATION.as_str();
+        const OCTET_STREAM: &str = names::OCTET_STREAM.as_str();
+        const JSON: &str = names::JSON.as_str();
+        const STAR: &str = names::_STAR.as_str();
+        const Q: &str = names::Q.as_str();
+
+        media_type_list.into_iter().for_each(|item| {
+            if let Ok(MediaType { ty, subty, suffix: _, params }) = item {
+                let q_accept = match (ty.as_str(), subty.as_str()) {
+                    (APPLICATION, OCTET_STREAM) => Some(Accept::Ssz),
+                    (APPLICATION, JSON) => Some(Accept::Json),
+                    (STAR, STAR) => Some(Accept::Any),
+                    _ => None,
+                }
+                .map(|item_accept_type| {
+                    let q_val = params
+                        .iter()
+                        .find_map(|(n, v)| match n.as_str() {
+                            Q => {
+                                Some((v.as_str().parse::<f32>().unwrap_or(0_f32) * 1000_f32) as u16)
+                            }
+                            _ => None,
+                        })
+                        .or(Some(1000_u16));
+
+                    (q_val.unwrap(), item_accept_type)
+                });
+
+                match q_accept {
+                    Some((q, accept)) if q > highest_q => {
+                        highest_q = q;
+                        accept_type = Some(accept);
+                    }
+                    _ => (),
+                }
+            }
+        });
+        accept_type.ok_or_else(|| "accept header is not supported".to_string())
+    }
+}
+
+#[must_use]
+#[derive(Debug, Clone, Copy, Default)]
+pub struct JsonOrSsz<T>(pub T);
+
+impl<T, S> FromRequest<S> for JsonOrSsz<T>
+where
+    T: serde::de::DeserializeOwned + ssz::Decode + 'static,
+    S: Send + Sync,
+{
+    type Rejection = Response;
+
+    async fn from_request(req: Request, _state: &S) -> Result<Self, Self::Rejection> {
+        let headers = req.headers().clone();
+        let content_type = headers.get(CONTENT_TYPE).and_then(|value| value.to_str().ok());
+
+        let bytes = Bytes::from_request(req, _state).await.map_err(IntoResponse::into_response)?;
+
+        if let Some(content_type) = content_type {
+            if content_type.starts_with(&ContentType::Json.to_string()) {
+                let payload: T = serde_json::from_slice(&bytes)
+                    .map_err(|_| StatusCode::BAD_REQUEST.into_response())?;
+                return Ok(Self(payload));
+            }
+
+            if content_type.starts_with(&ContentType::Ssz.to_string()) {
+                let payload = T::from_ssz_bytes(&bytes)
+                    .map_err(|_| StatusCode::BAD_REQUEST.into_response())?;
+                return Ok(Self(payload));
+            }
+        }
+
+        Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())
+    }
+}
+
 #[cfg(unix)]
 pub async fn wait_for_signal() -> eyre::Result<()> {
     use tokio::signal::unix::{SignalKind, signal};

crates/pbs/src/mev_boost/submit_block.rs

@@ -208,9 +208,9 @@ fn validate_unblinded_block_electra(
     let blobs = &block_response.blobs_bundle;
 
     let expected_commitments = &signed_blinded_block.body.blob_kzg_commitments;
-    if expected_commitments.len() != blobs.blobs.len() ||
-        expected_commitments.len() != blobs.commitments.len() ||
-        expected_commitments.len() != blobs.proofs.len()
+    if expected_commitments.len() != blobs.blobs.len()
+        || expected_commitments.len() != blobs.commitments.len()
+        || expected_commitments.len() != blobs.proofs.len()
     {
         return Err(PbsError::Validation(ValidationError::KzgCommitments {
             expected_blobs: expected_commitments.len(),

crates/pbs/src/routes/get_header.rs

@@ -41,7 +41,6 @@ pub async fn handle_get_header<S: BuilderApiState, A: BuilderApi<S>>(
         Ok(res) => {
             if let Some(max_bid) = res {
                 info!(value_eth = format_ether(max_bid.value()), block_hash =% max_bid.block_hash(), "received header");
-
                 BEACON_NODE_STATUS.with_label_values(&["200", GET_HEADER_ENDPOINT_TAG]).inc();
                 let response = match accept_header {
                     Accept::Ssz => {

tests/src/mock_validator.rs

@@ -115,6 +115,13 @@ impl MockValidator {
             ContentType::Ssz => signed_blinded_block.as_ssz_bytes(),
         };
 
+        let signed_blinded_block = signed_blinded_block_opt.unwrap_or_default();
+
+        let body = match content_type {
+            ContentType::Json => serde_json::to_vec(&signed_blinded_block).unwrap(),
+            ContentType::Ssz => signed_blinded_block.as_ssz_bytes(),
+        };
+
         Ok(self
             .comm_boost
             .client

tests/tests/pbs_get_header.rs

@@ -108,6 +108,50 @@ async fn test_get_header_ssz() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_get_header_ssz() -> Result<()> {
+    setup_test_env();
+    let signer = random_secret();
+    let pubkey: BlsPublicKey = blst_pubkey_to_alloy(&signer.sk_to_pk()).into();
+
+    let chain = Chain::Holesky;
+    let pbs_port = 3200;
+    let relay_port = pbs_port + 1;
+
+    // Run a mock relay
+    let mock_state = Arc::new(MockRelayState::new(chain, signer));
+    let mock_relay = generate_mock_relay(relay_port, *pubkey)?;
+    tokio::spawn(start_mock_relay_service(mock_state.clone(), relay_port));
+
+    // Run the PBS service
+    let config = to_pbs_config(chain, get_pbs_static_config(pbs_port), vec![mock_relay.clone()]);
+    let state = PbsState::new(config);
+    tokio::spawn(PbsService::run::<(), DefaultBuilderApi>(state));
+
+    // leave some time to start servers
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let mock_validator = MockValidator::new(pbs_port)?;
+    info!("Sending get header");
+    let res = mock_validator.do_get_header(None, Some(Accept::Ssz), ForkName::Electra).await?;
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let res: SignedExecutionPayloadHeader<ExecutionPayloadHeaderMessageElectra> =
+        SignedExecutionPayloadHeader::from_ssz_bytes(&res.bytes().await?).unwrap();
+
+    assert_eq!(mock_state.received_get_header(), 1);
+    assert_eq!(res.message.header.block_hash.0[0], 1);
+    assert_eq!(res.message.header.parent_hash, B256::ZERO);
+    assert_eq!(res.message.value, U256::from(10));
+    assert_eq!(res.message.pubkey, blst_pubkey_to_alloy(&mock_state.signer.sk_to_pk()));
+    assert_eq!(res.message.header.timestamp, timestamp_of_slot_start_sec(0, chain));
+    assert_eq!(
+        res.signature,
+        sign_builder_root(chain, &mock_state.signer, res.message.tree_hash_root().0)
+    );
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_get_header_returns_204_if_relay_down() -> Result<()> {
     setup_test_env();

tests/tests/pbs_post_blinded_blocks.rs

@@ -57,6 +57,47 @@ async fn test_submit_block_v2_ssz() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_submit_block_ssz() -> Result<()> {
+    setup_test_env();
+    let signer = random_secret();
+    let pubkey: BlsPublicKey = blst_pubkey_to_alloy(&signer.sk_to_pk()).into();
+
+    let chain = Chain::Holesky;
+    let pbs_port = 3800;
+
+    // Run a mock relay
+    let relays = vec![generate_mock_relay(pbs_port + 1, *pubkey)?];
+    let mock_state = Arc::new(MockRelayState::new(chain, signer));
+    tokio::spawn(start_mock_relay_service(mock_state.clone(), pbs_port + 1));
+
+    // Run the PBS service
+    let config = to_pbs_config(chain, get_pbs_static_config(pbs_port), relays);
+    let state = PbsState::new(config);
+    tokio::spawn(PbsService::run::<(), DefaultBuilderApi>(state));
+
+    // leave some time to start servers
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    let mock_validator = MockValidator::new(pbs_port)?;
+    info!("Sending submit block");
+    let res = mock_validator
+        .do_submit_block(
+            Some(SignedBlindedBeaconBlock::default()),
+            Accept::Ssz,
+            ContentType::Ssz,
+            ForkName::Electra,
+        )
+        .await?;
+
+    assert_eq!(res.status(), StatusCode::OK);
+    assert_eq!(mock_state.received_submit_block(), 1);
+
+    let response_body = PayloadAndBlobsElectra::from_ssz_bytes(&res.bytes().await?).unwrap();
+    assert_eq!(response_body.block_hash(), SubmitBlindedBlockResponse::default().block_hash());
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_submit_block_too_large() -> Result<()> {
     setup_test_env();