diff --git a/beacon_node/beacon_chain/src/test_utils.rs b/beacon_node/beacon_chain/src/test_utils.rs index 4fd3b4eb018..164a3c395c2 100644 --- a/beacon_node/beacon_chain/src/test_utils.rs +++ b/beacon_node/beacon_chain/src/test_utils.rs @@ -697,6 +697,12 @@ where let chain = builder.build().expect("should build"); + if execution_layer::test_utils::mock_rest_ssz_enabled() + && let Some(mock) = &self.mock_execution_layer + { + mock.el.spawn_watchdog_routine(chain.slot_clock.clone()); + } + BeaconChainHarness { spec: chain.spec.clone(), chain: Arc::new(chain), diff --git a/beacon_node/beacon_chain/tests/payload_invalidation.rs b/beacon_node/beacon_chain/tests/payload_invalidation.rs index c8dfefffca9..8f8f65d7069 100644 --- a/beacon_node/beacon_chain/tests/payload_invalidation.rs +++ b/beacon_node/beacon_chain/tests/payload_invalidation.rs @@ -10,10 +10,7 @@ use beacon_chain::{ canonical_head::CachedHead, test_utils::{BeaconChainHarness, EphemeralHarnessType, fork_name_from_env, test_spec}, }; -use execution_layer::{ - ExecutionLayer, ForkchoiceState, PayloadAttributes, - json_structures::{JsonForkchoiceStateV1, JsonPayloadAttributes, JsonPayloadAttributesV1}, -}; +use execution_layer::{ExecutionLayer, ForkchoiceState, PayloadAttributes}; use fork_choice::{Error as ForkChoiceError, InvalidationOperation, PayloadVerificationStatus}; use proto_array::{Error as ProtoArrayError, ExecutionStatus, ExecutionVerdict}; use slot_clock::SlotClock; @@ -126,24 +123,22 @@ impl InvalidPayloadRig { fn previous_forkchoice_update_params(&self) -> (ForkchoiceState, PayloadAttributes) { let mock_execution_layer = self.harness.mock_execution_layer.as_ref().unwrap(); - let json = mock_execution_layer + let (state, attributes) = mock_execution_layer .server - .take_previous_request() - .expect("no previous request"); - let params = json.get("params").expect("no params"); - - let fork_choice_state_json = params.get(0).expect("no payload param"); - let fork_choice_state: JsonForkchoiceStateV1 = - serde_json::from_value(fork_choice_state_json.clone()).unwrap(); - - let payload_param_json = params.get(1).expect("no payload param"); - let attributes: JsonPayloadAttributesV1 = - serde_json::from_value(payload_param_json.clone()).unwrap(); - - ( - fork_choice_state.into(), - JsonPayloadAttributes::V1(attributes).into(), - ) + .take_previous_forkchoice_request() + .expect("no previous forkchoice request"); + let attributes = attributes.expect("no payload attributes"); + let attributes = PayloadAttributes::new( + attributes.timestamp(), + attributes.prev_randao(), + attributes.suggested_fee_recipient(), + None, + None, + None, + None, + None, + ); + (state, attributes) } fn previous_payload_attributes(&self) -> PayloadAttributes { diff --git a/beacon_node/beacon_chain/tests/prepare_payload.rs b/beacon_node/beacon_chain/tests/prepare_payload.rs index 77c2db0e7cd..a10c30250db 100644 --- a/beacon_node/beacon_chain/tests/prepare_payload.rs +++ b/beacon_node/beacon_chain/tests/prepare_payload.rs @@ -12,8 +12,6 @@ use beacon_chain::{ }; use bls::{Keypair, Signature}; use eth2::types::{GraffitiPolicy, ProposerPreparationData}; -use execution_layer::http::{ENGINE_FORKCHOICE_UPDATED_V4, ENGINE_FORKCHOICE_UPDATED_V5}; -use execution_layer::json_structures::{JsonPayloadAttributesV4, JsonPayloadAttributesV5}; use execution_layer::{DEFAULT_GAS_LIMIT, PayloadAttributes}; use fork_choice::PayloadStatus; use logging::create_test_tracing_subscriber; @@ -896,32 +894,27 @@ async fn prepare_payload_around_heze_boundary(prepare_slot: Slot, heze_fork_epoc .await .unwrap(); - // Inspect the fcU request received by the mock execution layer - let request = harness + // Inspect the fcU request received by the mock execution layer. Captured by the shared mock + // core, so this covers both JSON-RPC and REST-SSZ. + let (_state, attributes) = harness .mock_execution_layer .as_ref() .unwrap() .server - .take_previous_request() - .expect("no previous request"); - let method = request.get("method").expect("no method"); - let params = request.get("params").expect("no params"); - let payload_attributes_json = params.get(1).expect("no payload attributes param"); + .take_previous_forkchoice_request() + .expect("no previous forkchoice request"); + let attributes = attributes.expect("no payload attributes"); if prepare_slot_is_heze { - assert_eq!(method, ENGINE_FORKCHOICE_UPDATED_V5); - let attributes: JsonPayloadAttributesV5 = - serde_json::from_value(payload_attributes_json.clone()).unwrap(); + let PayloadAttributes::V5(attributes) = &attributes else { + panic!("expected V5 payload attributes at heze, got {attributes:?}"); + }; // We are currently sending the V5 shape with an empty inclusion list assert!(attributes.inclusion_list_transactions.is_empty()); } else { - assert_eq!(method, ENGINE_FORKCHOICE_UPDATED_V4); - let _attributes: JsonPayloadAttributesV4 = - serde_json::from_value(payload_attributes_json.clone()).unwrap(); assert!( - payload_attributes_json - .get("inclusionListTransactions") - .is_none() + matches!(attributes, PayloadAttributes::V4(_)), + "expected V4 payload attributes before heze, got {attributes:?}" ); } diff --git a/beacon_node/execution_layer/src/engine_api/json_structures.rs b/beacon_node/execution_layer/src/engine_api/json_structures.rs index e594f996155..007cda30abe 100644 --- a/beacon_node/execution_layer/src/engine_api/json_structures.rs +++ b/beacon_node/execution_layer/src/engine_api/json_structures.rs @@ -699,7 +699,7 @@ impl TryFrom for ExecutionRequestsGloas { partial_getter_error(ty = "Error", expr = "Error::IncorrectStateVariant") )] #[derive(Debug, PartialEq, Serialize, Deserialize)] -#[serde(untagged)] +#[serde(bound = "E: EthSpec", untagged)] pub struct JsonGetPayloadResponse { #[superstruct( only(Bellatrix), @@ -728,6 +728,71 @@ pub struct JsonGetPayloadResponse { pub execution_requests: JsonExecutionRequests, } +impl TryFrom> for JsonGetPayloadResponse { + type Error = ssz_types::Error; + + fn try_from(response: GetPayloadResponse) -> Result { + match response { + GetPayloadResponse::Bellatrix(inner) => Ok(JsonGetPayloadResponse::Bellatrix( + JsonGetPayloadResponseBellatrix { + execution_payload: inner.execution_payload.into(), + block_value: inner.block_value, + }, + )), + GetPayloadResponse::Capella(inner) => Ok(JsonGetPayloadResponse::Capella( + JsonGetPayloadResponseCapella { + execution_payload: inner.execution_payload.try_into()?, + block_value: inner.block_value, + }, + )), + GetPayloadResponse::Deneb(inner) => { + Ok(JsonGetPayloadResponse::Deneb(JsonGetPayloadResponseDeneb { + execution_payload: inner.execution_payload.try_into()?, + block_value: inner.block_value, + blobs_bundle: inner.blobs_bundle.into(), + should_override_builder: inner.should_override_builder, + })) + } + GetPayloadResponse::Electra(inner) => Ok(JsonGetPayloadResponse::Electra( + JsonGetPayloadResponseElectra { + execution_payload: inner.execution_payload.try_into()?, + block_value: inner.block_value, + blobs_bundle: inner.blobs_bundle.into(), + should_override_builder: inner.should_override_builder, + execution_requests: ExecutionRequests::Electra(inner.requests).into(), + }, + )), + GetPayloadResponse::Fulu(inner) => { + Ok(JsonGetPayloadResponse::Fulu(JsonGetPayloadResponseFulu { + execution_payload: inner.execution_payload.try_into()?, + block_value: inner.block_value, + blobs_bundle: inner.blobs_bundle.into(), + should_override_builder: inner.should_override_builder, + execution_requests: ExecutionRequests::Electra(inner.requests).into(), + })) + } + GetPayloadResponse::Gloas(inner) => { + Ok(JsonGetPayloadResponse::Gloas(JsonGetPayloadResponseGloas { + execution_payload: inner.execution_payload.try_into()?, + block_value: inner.block_value, + blobs_bundle: inner.blobs_bundle.into(), + should_override_builder: inner.should_override_builder, + execution_requests: ExecutionRequests::Gloas(inner.requests).into(), + })) + } + GetPayloadResponse::Heze(inner) => { + Ok(JsonGetPayloadResponse::Heze(JsonGetPayloadResponseHeze { + execution_payload: inner.execution_payload.try_into()?, + block_value: inner.block_value, + blobs_bundle: inner.blobs_bundle.into(), + should_override_builder: inner.should_override_builder, + execution_requests: ExecutionRequests::Gloas(inner.requests).into(), + })) + } + } + } +} + impl TryFrom> for GetPayloadResponse { type Error = String; fn try_from(json_get_payload_response: JsonGetPayloadResponse) -> Result { diff --git a/beacon_node/execution_layer/src/engine_api/rest.rs b/beacon_node/execution_layer/src/engine_api/rest.rs index e247168627d..e4039af9d7c 100644 --- a/beacon_node/execution_layer/src/engine_api/rest.rs +++ b/beacon_node/execution_layer/src/engine_api/rest.rs @@ -52,7 +52,7 @@ pub enum HttpVersion { } impl HttpVersion { - pub fn as_str(&self) -> &'static str { + fn as_str(&self) -> &'static str { match self { HttpVersion::Http2 => "http2", HttpVersion::Http1 => "http1", @@ -615,14 +615,16 @@ pub fn header_to_fork(header: &str) -> Option { mod tests { use super::*; use crate::auth::JwtKey; - use crate::engine_api::{NewPayloadRequestDeneb, NewPayloadRequestGloas}; + use crate::engine_api::{ + NewPayloadRequestDeneb, NewPayloadRequestFulu, NewPayloadRequestGloas, + }; use crate::test_utils::{Config, DEFAULT_JWT_SECRET, MockExecutionConfig, MockServer}; use std::future::Future; use std::sync::Arc; use tokio::runtime; use types::{ - Address, ExecutionPayloadDeneb, ExecutionPayloadGloas, ExecutionRequestsGloas, - MainnetEthSpec, + Address, ExecutionPayloadDeneb, ExecutionPayloadFulu, ExecutionPayloadGloas, + ExecutionRequestsElectra, ExecutionRequestsGloas, MainnetEthSpec, }; #[test] @@ -795,6 +797,19 @@ mod tests { ); self } + + /// Asserts the typed value the client decoded from the mock's SSZ response. + async fn assert_ssz_response(self, request_func: R, check: impl FnOnce(T)) -> Self + where + R: Fn(Arc) -> F, + F: Future>, + { + let value = request_func(self.client.clone()) + .await + .expect("rest request should succeed"); + check(value); + self + } } #[tokio::test] @@ -1184,4 +1199,185 @@ mod tests { ) .await; } + + #[tokio::test] + async fn new_payload_valid_round_trip() { + let payload = ExecutionPayloadFulu::::default(); + let execution_requests = ExecutionRequestsElectra::::default(); + let tester = RestTester::new(true); + tester.server.all_payloads_valid_on_new_payload(); + tester + .assert_ssz_response( + move |client| { + let payload = payload.clone(); + let execution_requests = execution_requests.clone(); + async move { + let request = NewPayloadRequest::Fulu(NewPayloadRequestFulu { + execution_payload: &payload, + versioned_hashes: vec![], + parent_beacon_block_root: Hash256::repeat_byte(0), + execution_requests: &execution_requests, + }); + client.new_payload::(request).await + } + }, + |status| assert_eq!(status.status, PayloadStatusV1Status::Valid), + ) + .await; + } + + #[tokio::test] + async fn new_payload_syncing_round_trip() { + let payload = ExecutionPayloadFulu::::default(); + let execution_requests = ExecutionRequestsElectra::::default(); + let tester = RestTester::new(true); + tester.server.all_payloads_syncing_on_new_payload(false); + tester + .assert_ssz_response( + move |client| { + let payload = payload.clone(); + let execution_requests = execution_requests.clone(); + async move { + let request = NewPayloadRequest::Fulu(NewPayloadRequestFulu { + execution_payload: &payload, + versioned_hashes: vec![], + parent_beacon_block_root: Hash256::repeat_byte(0), + execution_requests: &execution_requests, + }); + client.new_payload::(request).await + } + }, + |status| assert_eq!(status.status, PayloadStatusV1Status::Syncing), + ) + .await; + } + + #[tokio::test] + async fn new_payload_invalid_block_hash_folds_to_invalid_round_trip() { + let payload = ExecutionPayloadFulu::::default(); + let execution_requests = ExecutionRequestsElectra::::default(); + let tester = RestTester::new(true); + tester + .server + .all_payloads_invalid_block_hash_on_new_payload(); + tester + .assert_ssz_response( + move |client| { + let payload = payload.clone(); + let execution_requests = execution_requests.clone(); + async move { + let request = NewPayloadRequest::Fulu(NewPayloadRequestFulu { + execution_payload: &payload, + versioned_hashes: vec![], + parent_beacon_block_root: Hash256::repeat_byte(0), + execution_requests: &execution_requests, + }); + client.new_payload::(request).await + } + }, + |status| assert_eq!(status.status, PayloadStatusV1Status::Invalid), + ) + .await; + } + + #[tokio::test] + async fn forkchoice_updated_no_attrs_round_trip() { + let state = forkchoice_state(); + let tester = RestTester::new(true); + tester.server.set_fcu_payload_status( + state.head_block_hash, + PayloadStatusV1 { + status: PayloadStatusV1Status::Valid, + latest_valid_hash: None, + validation_error: None, + inclusion_list_satisfied: None, + }, + ); + tester + .assert_ssz_response( + move |client| async move { + client + .forkchoice_updated::(ForkName::Fulu, state, None, None) + .await + }, + |response| { + assert_eq!(response.payload_status.status, PayloadStatusV1Status::Valid); + assert!(response.payload_id.is_none()); + }, + ) + .await; + } + + #[tokio::test] + async fn forkchoice_updated_with_attrs_round_trip() { + let state = forkchoice_state(); + let attributes = PayloadAttributes::new( + 1_000, + Hash256::repeat_byte(4), + Address::repeat_byte(9), + Some(vec![]), + Some(Hash256::repeat_byte(5)), + None, + None, + None, + ); + let tester = RestTester::new(true); + tester.server.set_fcu_payload_status( + state.head_block_hash, + PayloadStatusV1 { + status: PayloadStatusV1Status::Valid, + latest_valid_hash: None, + validation_error: None, + inclusion_list_satisfied: None, + }, + ); + tester + .assert_ssz_response( + move |client| { + let attributes = attributes.clone(); + async move { + client + .forkchoice_updated::( + ForkName::Fulu, + state, + Some(attributes), + None, + ) + .await + } + }, + |response| { + assert_eq!(response.payload_status.status, PayloadStatusV1Status::Valid); + }, + ) + .await; + } + + #[tokio::test] + async fn get_payload_unknown_id_returns_404() { + let tester = RestTester::new(true); + let payload_id: PayloadId = [9; 8]; + let result = tester + .client + .get_payload::(ForkName::Fulu, payload_id) + .await; + assert!( + matches!(&result, Err(e) if e.is_unknown_payload()), + "expected unknown-payload error, got {result:?}" + ); + } + + #[tokio::test] + async fn get_blobs_v2_miss_returns_none() { + RestTester::new(true) + .assert_ssz_response( + |client| async move { + client + .get_blobs_v2::(vec![Hash256::repeat_byte(1)]) + .await + }, + |blobs| assert!(blobs.is_none()), + ) + .await; + } } diff --git a/beacon_node/execution_layer/src/lib.rs b/beacon_node/execution_layer/src/lib.rs index 0165358b427..a4fea0a55f8 100644 --- a/beacon_node/execution_layer/src/lib.rs +++ b/beacon_node/execution_layer/src/lib.rs @@ -17,7 +17,9 @@ pub use engine_api::EngineCapabilities; use engine_api::Error as ApiError; use engine_api::transport::EngineApi; pub use engine_api::*; -pub use engine_api::{http, http::HttpJsonRpc, http::deposit_methods, rest::HttpRestSsz}; +pub use engine_api::{ + http, http::HttpJsonRpc, http::deposit_methods, rest::HttpRestSsz, transport::Transport, +}; use engines::{Engine, EngineError}; pub use engines::{EngineState, ForkchoiceState}; use eth2::types::{BlobsBundle, FullPayloadContents}; @@ -1984,6 +1986,10 @@ impl ExecutionLayer { Err(Error::NoPayloadBuilder) } } + + pub(crate) fn resolved_transport(&self) -> Option { + self.engine().api.get_decision().copied() + } } #[derive(AsRefStr)] @@ -2207,6 +2213,8 @@ mod test { async fn produce_three_valid_pos_execution_blocks() { let runtime = TestRuntime::default(); MockExecutionLayer::default_params(runtime.task_executor.clone()) + .resolve_and_assert_transport() + .await .produce_valid_execution_payload_on_head() .await .produce_valid_execution_payload_on_head() diff --git a/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs b/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs index 914c1dffd76..21dda50e6ee 100644 --- a/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs +++ b/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs @@ -494,6 +494,10 @@ impl ExecutionBlockGenerator { self.payload_ids.get(id).cloned() } + pub fn expire_all_payload_ids(&mut self) { + self.payload_ids.clear(); + } + pub fn get_blobs_bundle(&mut self, id: &PayloadId) -> Option> { self.blobs_bundles.get(id).cloned() } diff --git a/beacon_node/execution_layer/src/test_utils/handle_rest.rs b/beacon_node/execution_layer/src/test_utils/handle_rest.rs new file mode 100644 index 00000000000..71d7c761f50 --- /dev/null +++ b/beacon_node/execution_layer/src/test_utils/handle_rest.rs @@ -0,0 +1,625 @@ +use bytes::Bytes; +use serde_json::json; +use ssz::{Decode, DecodeError, Encode}; +use ssz_types::{ProgressiveVariableList, VariableList}; +use types::{ + EthSpec, ExecutionBlockHash, ExecutionPayload, ForkName, Hash256, ProgressiveTransactions, + ProgressiveWithdrawals, +}; +use warp::http::Response; + +use crate::engine_api::rest::header_to_fork; +use crate::engine_api::{ExecutionPayloadBodyV2, ForkchoiceUpdatedResponse, PayloadAttributes}; +use crate::engines::ForkchoiceState; +use crate::json_structures::{BlobAndProof, BlobAndProofV2, BlobCellsAndProofsV1, JsonCell}; +use crate::ssz_structures::*; + +use super::handle_rpc::UNKNOWN_PAYLOAD_ERROR_CODE; +use super::mock_engine_core::CorePayload; +use super::{Context, DEFAULT_CLIENT_VERSION, STUB_CAPABILITIES_JSON}; + +pub(crate) fn handle_rest( + method: &str, + path: &str, + query: Option<&str>, + eth_execution_version: Option<&str>, + body: &Bytes, + ctx: &Context, +) -> Response { + match (method, path) { + ("GET", "/engine/v1/capabilities") => { + let mut capabilities: serde_json::Value = + serde_json::from_str(STUB_CAPABILITIES_JSON).unwrap(); + if !ctx.engine_capabilities.read().get_client_version_v1() + && let Some(endpoints) = capabilities["unscoped_endpoints"].as_array_mut() + { + endpoints.retain(|e| e.as_str() != Some("identity")); + } + json_response(Bytes::from(serde_json::to_vec(&capabilities).unwrap())) + } + ("GET", "/engine/v1/identity") => json_response(Bytes::from( + serde_json::to_vec(std::slice::from_ref(&*DEFAULT_CLIENT_VERSION)).unwrap(), + )), + ("POST", "/engine/v1/payloads") => { + let Some(fork) = eth_execution_version.and_then(header_to_fork) else { + return problem_response( + RestProblemKind::UnsupportedFork, + eth_execution_version.map(|header| header.to_string()), + ); + }; + + let Ok(payload) = decode_new_payload::(body, fork) else { + return problem_response(RestProblemKind::SszDecodeError, None); + }; + + let status = match ctx.core_new_payload(payload) { + Ok(status) => status, + Err(err) => return core_error_response(err), + }; + + match SszPayloadStatusV1::::try_from(status) { + Ok(ssz) => ssz_response(Bytes::from(ssz.as_ssz_bytes())), + Err(e) => problem_response(RestProblemKind::Internal, Some(format!("{e:?}"))), + } + } + ("POST", "/engine/v1/forkchoice") => { + let Some(fork) = eth_execution_version.and_then(header_to_fork) else { + return problem_response( + RestProblemKind::UnsupportedFork, + eth_execution_version.map(|header| header.to_string()), + ); + }; + + let Ok((state, attributes)) = decode_forkchoice_updated::(body, fork) else { + return problem_response(RestProblemKind::SszDecodeError, None); + }; + + let response = match ctx.core_forkchoice_updated(state, attributes) { + Ok(response) => response, + Err(err) => return core_error_response(err), + }; + + match SszForkchoiceUpdatedResponse::::try_from(ForkchoiceUpdatedResponse::from( + response, + )) { + Ok(ssz) => ssz_response(Bytes::from(ssz.as_ssz_bytes())), + Err(e) => problem_response(RestProblemKind::Internal, Some(format!("{e:?}"))), + } + } + ("GET", path) if path.starts_with("/engine/v1/payloads/") => { + if eth_execution_version.and_then(header_to_fork).is_none() { + return problem_response( + RestProblemKind::UnsupportedFork, + eth_execution_version.map(|header| header.to_string()), + ); + } + + let Some(payload_id) = path + .strip_prefix("/engine/v1/payloads/") + .and_then(|segment| segment.strip_prefix("0x")) + .and_then(|hex| hex::decode(hex).ok()) + .and_then(|bytes| <[u8; 8]>::try_from(bytes).ok()) + else { + return problem_response( + RestProblemKind::InvalidRequest, + Some(format!("malformed payload id: {path}")), + ); + }; + + let core = match ctx.core_get_payload(payload_id) { + Ok(core) => core, + Err(err) => return core_error_response(err), + }; + + match encode_get_payload(core) { + Ok(bytes) => ssz_response_no_store(Bytes::from(bytes)), + Err(e) => problem_response(RestProblemKind::Internal, Some(e)), + } + } + ("POST", "/engine/v1/blobs/v2") => handle_get_blobs(ctx, body, true), + ("POST", "/engine/v1/blobs/v3") => handle_get_blobs(ctx, body, false), + ("POST", "/engine/v1/blobs/v4") => handle_get_blobs_v4(ctx, body), + ("POST", "/engine/v1/bodies/hash") => { + let Some(fork) = eth_execution_version.and_then(header_to_fork) else { + return problem_response( + RestProblemKind::UnsupportedFork, + eth_execution_version.map(|header| header.to_string()), + ); + }; + + let Ok(block_hashes) = decode_bodies_by_hash_request(body) else { + return problem_response(RestProblemKind::SszDecodeError, None); + }; + + let generator = ctx.execution_block_generator.read(); + let payloads = block_hashes + .into_iter() + .map(|hash| { + generator.execution_payload_by_hash(ExecutionBlockHash::from_root(hash)) + }) + .collect(); + drop(generator); + + match encode_bodies_response(payloads, fork) { + Ok(bytes) => ssz_response(Bytes::from(bytes)), + Err(e) => problem_response(RestProblemKind::Internal, Some(e)), + } + } + ("GET", "/engine/v1/bodies") => { + let Some(fork) = eth_execution_version.and_then(header_to_fork) else { + return problem_response( + RestProblemKind::UnsupportedFork, + eth_execution_version.map(|header| header.to_string()), + ); + }; + + let Some((from, count)) = parse_from_count(query) else { + return problem_response( + RestProblemKind::InvalidRequest, + Some("malformed bodies range query".to_string()), + ); + }; + + let generator = ctx.execution_block_generator.read(); + let payloads = (from..from.saturating_add(count)) + .map(|number| generator.execution_payload_by_number(number)) + .collect(); + drop(generator); + + match encode_bodies_response(payloads, fork) { + Ok(bytes) => ssz_response(Bytes::from(bytes)), + Err(e) => problem_response(RestProblemKind::Internal, Some(e)), + } + } + _ => problem_response(RestProblemKind::MethodNotFound, None), + } +} + +fn json_response(body: Bytes) -> Response { + Response::builder() + .status(200) + .header("Content-Type", "application/json") + .body(body) + .unwrap() +} + +fn ssz_response(body: Bytes) -> Response { + Response::builder() + .status(200) + .header("Content-Type", "application/octet-stream") + .body(body) + .unwrap() +} + +fn ssz_response_no_store(body: Bytes) -> Response { + Response::builder() + .status(200) + .header("Content-Type", "application/octet-stream") + .header("Cache-Control", "no-store") + .body(body) + .unwrap() +} + +fn decode_new_payload( + body: &[u8], + fork: ForkName, +) -> Result, DecodeError> { + Ok(match fork { + ForkName::Bellatrix => ExecutionPayload::Bellatrix( + SszExecutionPayloadEnvelopeBellatrix::from_ssz_bytes(body)?.execution_payload, + ), + ForkName::Capella => ExecutionPayload::Capella( + SszExecutionPayloadEnvelopeCapella::from_ssz_bytes(body)?.execution_payload, + ), + ForkName::Deneb => ExecutionPayload::Deneb( + SszExecutionPayloadEnvelopeDeneb::from_ssz_bytes(body)?.execution_payload, + ), + ForkName::Electra => ExecutionPayload::Electra( + SszExecutionPayloadEnvelopeElectra::from_ssz_bytes(body)?.execution_payload, + ), + ForkName::Fulu => ExecutionPayload::Fulu( + SszExecutionPayloadEnvelopeFulu::from_ssz_bytes(body)?.execution_payload, + ), + ForkName::Gloas => ExecutionPayload::Gloas( + SszExecutionPayloadEnvelopeGloas::from_ssz_bytes(body)?.execution_payload, + ), + ForkName::Heze => ExecutionPayload::Heze( + SszExecutionPayloadEnvelopeHeze::from_ssz_bytes(body)?.execution_payload, + ), + ForkName::Base | ForkName::Altair => { + return Err(DecodeError::BytesInvalid(format!( + "unsupported fork for new_payload envelope: {fork}" + ))); + } + }) +} + +fn decode_forkchoice_updated( + body: &[u8], + fork: ForkName, +) -> Result<(ForkchoiceState, Option), DecodeError> { + Ok(match fork { + ForkName::Bellatrix => { + let update = SszForkchoiceUpdateBellatrix::from_ssz_bytes(body)?; + let attributes = update.payload_attributes.first().cloned(); + ( + update.forkchoice_state, + attributes.map(PayloadAttributes::V1), + ) + } + ForkName::Capella => { + let update = SszForkchoiceUpdateCapella::from_ssz_bytes(body)?; + let attributes = update.payload_attributes.first().cloned(); + ( + update.forkchoice_state, + attributes.map(PayloadAttributes::V2), + ) + } + ForkName::Deneb => { + let update = SszForkchoiceUpdateDeneb::from_ssz_bytes(body)?; + let attributes = update.payload_attributes.first().cloned(); + ( + update.forkchoice_state, + attributes.map(PayloadAttributes::V3), + ) + } + ForkName::Electra => { + let update = SszForkchoiceUpdateElectra::from_ssz_bytes(body)?; + let attributes = update.payload_attributes.first().cloned(); + ( + update.forkchoice_state, + attributes.map(PayloadAttributes::V3), + ) + } + ForkName::Fulu => { + let update = SszForkchoiceUpdateFulu::from_ssz_bytes(body)?; + let attributes = update.payload_attributes.first().cloned(); + ( + update.forkchoice_state, + attributes.map(PayloadAttributes::V3), + ) + } + ForkName::Gloas => { + let update = SszForkchoiceUpdateCustodyColumnsGloas::::from_ssz_bytes(body)?; + let attributes = update.payload_attributes.first().cloned(); + ( + update.forkchoice_state, + attributes.map(PayloadAttributes::V4), + ) + } + ForkName::Heze => { + let update = SszForkchoiceUpdateCustodyColumnsHeze::::from_ssz_bytes(body)?; + let attributes = update.payload_attributes.first().cloned(); + ( + update.forkchoice_state, + attributes.map(PayloadAttributes::V5), + ) + } + ForkName::Base | ForkName::Altair => { + return Err(DecodeError::BytesInvalid(format!( + "unsupported fork for forkchoice update: {fork}" + ))); + } + }) +} + +fn encode_get_payload(core: CorePayload) -> Result, String> { + Ok( + SszGetPayloadResponse::try_from(core.into_get_payload_response()) + .map_err(|e| format!("{e:?}"))? + .as_ssz_bytes(), + ) +} + +fn decode_blobs_request(body: &[u8]) -> Result, DecodeError> { + Ok(SszBlobsRequestV1::::from_ssz_bytes(body)? + .versioned_hashes + .to_vec()) +} + +fn no_content_response() -> Response { + Response::builder().status(204).body(Bytes::new()).unwrap() +} + +/// `all_or_nothing = true` for `/blobs/v2` (any miss ⇒ `204`), `false` for `/blobs/v3` (partial). +fn handle_get_blobs( + ctx: &Context, + body: &[u8], + all_or_nothing: bool, +) -> Response { + let Ok(versioned_hashes) = decode_blobs_request::(body) else { + return problem_response(RestProblemKind::SszDecodeError, None); + }; + + let generator = ctx.execution_block_generator.read(); + let mut entries = Vec::with_capacity(versioned_hashes.len()); + let mut all_present = true; + for hash in &versioned_hashes { + match generator.get_blob_and_proof(hash) { + Some(BlobAndProof::V2(contents)) => entries.push(BlobsEntryV2 { + available: true, + contents, + }), + _ => { + all_present = false; + entries.push(BlobsEntryV2 { + available: false, + contents: BlobAndProofV2 { + blob: Default::default(), + proofs: Default::default(), + }, + }); + } + } + } + + if all_or_nothing && !all_present { + return no_content_response(); + } + + let entries = match VariableList::new(entries) { + Ok(entries) => entries, + Err(e) => return problem_response(RestProblemKind::Internal, Some(format!("{e:?}"))), + }; + ssz_response(Bytes::from(SszBlobsResponseV2 { entries }.as_ssz_bytes())) +} + +/// `/blobs/v4` serves each blob's requested-column cells + proofs, with a blob the mock +/// doesn't have marked `available = false`. +fn handle_get_blobs_v4(ctx: &Context, body: &[u8]) -> Response { + let Ok(request) = SszBlobsRequestV2::::from_ssz_bytes(body) else { + return problem_response(RestProblemKind::SszDecodeError, None); + }; + let versioned_hashes = request.versioned_hashes.to_vec(); + let requested_indices: Vec = request + .indices_bitarray + .iter() + .enumerate() + .filter_map(|(index, set)| set.then_some(index)) + .collect(); + + let generator = ctx.execution_block_generator.read(); + let mut entries = Vec::with_capacity(versioned_hashes.len()); + for hash in &versioned_hashes { + let entry = match generator.get_blob_cells_and_proofs(hash) { + Some((cells, cell_proofs)) => { + let mut blob_cells = Vec::with_capacity(requested_indices.len()); + let mut proofs = Vec::with_capacity(requested_indices.len()); + for &index in &requested_indices { + let Some((cell, proof)) = cells.get(index).zip(cell_proofs.get(index)) else { + return problem_response( + RestProblemKind::Internal, + Some(format!("cell index {index} out of range")), + ); + }; + blob_cells.push(Some(JsonCell(cell.clone()))); + proofs.push(Some(*proof)); + } + let contents = match SszBlobCellsAndProofs::try_from(BlobCellsAndProofsV1:: { + blob_cells, + proofs, + }) { + Ok(contents) => contents, + Err(e) => { + return problem_response(RestProblemKind::Internal, Some(format!("{e:?}"))); + } + }; + BlobsEntryV4 { + available: true, + contents, + } + } + None => BlobsEntryV4 { + available: false, + contents: SszBlobCellsAndProofs { + blob_cells: VariableList::empty(), + proofs: VariableList::empty(), + }, + }, + }; + entries.push(entry); + } + + let entries = match VariableList::new(entries) { + Ok(entries) => entries, + Err(e) => return problem_response(RestProblemKind::Internal, Some(format!("{e:?}"))), + }; + ssz_response(Bytes::from(SszBlobsResponseV4 { entries }.as_ssz_bytes())) +} + +fn encode_bodies_response( + payloads: Vec>>, + fork: ForkName, +) -> Result, String> { + let bodies: Vec>> = payloads + .into_iter() + .map(|maybe_payload| { + maybe_payload + .map(|payload| -> Result, String> { + let transactions = payload + .transactions() + .iter() + .map(|transaction| ProgressiveVariableList::new(transaction.to_vec())) + .collect::, _>>() + .and_then(ProgressiveTransactions::new) + .map_err(|e| { + format!("failed to build progressive transactions list: {e:?}") + })?; + let withdrawals = payload + .withdrawals() + .ok() + .map(|withdrawals| ProgressiveWithdrawals::::new(withdrawals.to_vec())) + .transpose() + .map_err(|e| format!("withdrawals exceed SSZ bound: {e:?}"))?; + Ok(ExecutionPayloadBodyV2 { + transactions, + withdrawals, + block_access_list: payload.block_access_list().ok().cloned(), + }) + }) + .transpose() + }) + .collect::, String>>()?; + + let bytes = match fork { + ForkName::Bellatrix => { + let entries = bodies + .into_iter() + .map(|body| { + Ok(match body { + Some(body) => SszBodyEntryV1 { + available: true, + body: SszExecutionPayloadBodyV1::try_from(body) + .map_err(|e| format!("{e:?}"))?, + }, + None => SszBodyEntryV1 { + available: false, + body: SszExecutionPayloadBodyV1 { + transactions: VariableList::empty(), + }, + }, + }) + }) + .collect::>, String>>()?; + SszBodiesResponseV1 { + entries: VariableList::new(entries).map_err(|e| format!("{e:?}"))?, + } + .as_ssz_bytes() + } + ForkName::Capella | ForkName::Deneb | ForkName::Electra | ForkName::Fulu => { + let entries = bodies + .into_iter() + .map(|body| { + Ok(match body { + Some(body) => SszBodyEntryV2 { + available: true, + body: SszExecutionPayloadBodyV2::try_from(body) + .map_err(|e| format!("{e:?}"))?, + }, + None => SszBodyEntryV2 { + available: false, + body: SszExecutionPayloadBodyV2 { + transactions: VariableList::empty(), + withdrawals: VariableList::empty(), + }, + }, + }) + }) + .collect::>, String>>()?; + SszBodiesResponseV2 { + entries: VariableList::new(entries).map_err(|e| format!("{e:?}"))?, + } + .as_ssz_bytes() + } + ForkName::Gloas | ForkName::Heze => { + let entries = bodies + .into_iter() + .map(|body| { + Ok(match body { + Some(body) => SszBodyEntryV3 { + available: true, + body: SszExecutionPayloadBodyV3::try_from(body) + .map_err(|e| format!("{e:?}"))?, + }, + None => SszBodyEntryV3 { + available: false, + body: SszExecutionPayloadBodyV3 { + transactions: VariableList::empty(), + withdrawals: VariableList::empty(), + block_access_list: VariableList::empty(), + }, + }, + }) + }) + .collect::>, String>>()?; + SszBodiesResponseV3 { + entries: VariableList::new(entries).map_err(|e| format!("{e:?}"))?, + } + .as_ssz_bytes() + } + ForkName::Base | ForkName::Altair => { + return Err(format!( + "unsupported fork for execution payload bodies: {fork}" + )); + } + }; + + Ok(bytes) +} + +fn decode_bodies_by_hash_request(body: &[u8]) -> Result, DecodeError> { + Ok(SszBodiesByHashRequest::from_ssz_bytes(body)? + .block_hashes + .to_vec()) +} + +fn parse_from_count(query: Option<&str>) -> Option<(u64, u64)> { + let query = query?; + let mut from = None; + let mut count = None; + for pair in query.split('&') { + let (key, value) = pair.split_once('=')?; + match key { + "from" => from = value.parse().ok(), + "count" => count = value.parse().ok(), + _ => {} + } + } + Some((from?, count?)) +} + +pub enum RestProblemKind { + UnsupportedFork, + SszDecodeError, + InvalidRequest, + MethodNotFound, + UnknownPayload, + Internal, +} + +impl RestProblemKind { + fn status(&self) -> u16 { + match self { + RestProblemKind::UnsupportedFork => 400, + RestProblemKind::SszDecodeError => 400, + RestProblemKind::InvalidRequest => 400, + RestProblemKind::MethodNotFound => 404, + RestProblemKind::UnknownPayload => 404, + RestProblemKind::Internal => 500, + } + } + + fn slug(&self) -> &'static str { + match self { + RestProblemKind::UnsupportedFork => "unsupported-fork", + RestProblemKind::SszDecodeError => "ssz-decode-error", + RestProblemKind::InvalidRequest => "invalid-request", + RestProblemKind::MethodNotFound => "method-not-found", + RestProblemKind::UnknownPayload => "unknown-payload", + RestProblemKind::Internal => "internal", + } + } +} + +pub fn problem_response(problem: RestProblemKind, detail: Option) -> Response { + let body = json!({ + "type": format!("/engine-api/errors/{}", problem.slug()), + "detail": detail, + }); + + Response::builder() + .status(problem.status()) + .header("Content-Type", "application/problem+json") + .body(Bytes::from(serde_json::to_vec(&body).unwrap())) + .unwrap() +} + +pub fn core_error_response((message, code): (String, i64)) -> Response { + let problem = if code == UNKNOWN_PAYLOAD_ERROR_CODE { + RestProblemKind::UnknownPayload + } else { + RestProblemKind::Internal + }; + + problem_response(problem, Some(message)) +} diff --git a/beacon_node/execution_layer/src/test_utils/handle_rpc.rs b/beacon_node/execution_layer/src/test_utils/handle_rpc.rs index 9e443ae30a8..6e3ecbafa2d 100644 --- a/beacon_node/execution_layer/src/test_utils/handle_rpc.rs +++ b/beacon_node/execution_layer/src/test_utils/handle_rpc.rs @@ -1,7 +1,7 @@ use super::Context; use crate::engine_api::{http::*, *}; use crate::json_structures::*; -use crate::test_utils::{DEFAULT_CLIENT_VERSION, DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI}; +use crate::test_utils::DEFAULT_CLIENT_VERSION; use serde::de::DeserializeOwned; use serde_json::Value as JsonValue; use std::sync::Arc; @@ -221,45 +221,17 @@ pub async fn handle_rpc( _ => unreachable!(), }; - // Canned responses set by block hash take priority. - if let Some(status) = ctx.get_new_payload_status(request.block_hash()) { - return status - .map(|status| serde_json::to_value(JsonPayloadStatusV1::from(status)).unwrap()) - .map_err(|message| (message, GENERIC_ERROR_CODE)); - } - - let (static_response, should_import) = - if let Some(mut response) = ctx.static_new_payload_response.lock().clone() { - if response.status.status == PayloadStatusV1Status::Valid { - response.status.latest_valid_hash = Some(*request.block_hash()) - } - - (Some(response.status), response.should_import) - } else { - (None, true) - }; - - let dynamic_response = if should_import { - Some( - ctx.execution_block_generator - .write() - .new_payload(request.try_into().unwrap()), - ) - } else { - None - }; - - let mut response = static_response.or(dynamic_response).unwrap(); + let mut status = ctx.core_new_payload(request.try_into().unwrap())?; if method == ENGINE_NEW_PAYLOAD_V6 { // TODO(heze): make this configurable so enforcement tests can exercise an // unsatisfied payload. - if response.status == PayloadStatusV1Status::Valid { - response.inclusion_list_satisfied = Some(true); + if status.status == PayloadStatusV1Status::Valid { + status.inclusion_list_satisfied = Some(true); } - Ok(serde_json::to_value(JsonPayloadStatusV2::from(response)).unwrap()) + Ok(serde_json::to_value(JsonPayloadStatusV2::from(status)).unwrap()) } else { - Ok(serde_json::to_value(JsonPayloadStatusV1::from(response)).unwrap()) + Ok(serde_json::to_value(JsonPayloadStatusV1::from(status)).unwrap()) } } ENGINE_GET_PAYLOAD_V1 @@ -272,42 +244,18 @@ pub async fn handle_rpc( get_param(params, 0).map_err(|s| (s, BAD_PARAMS_ERROR_CODE))?; let id = request.into(); - let response = ctx - .execution_block_generator - .write() - .get_payload(&id) - .ok_or_else(|| { - ( - format!("no payload for id {:?}", id), - UNKNOWN_PAYLOAD_ERROR_CODE, - ) - })?; - - let maybe_blobs = ctx.execution_block_generator.write().get_blobs_bundle(&id); - let maybe_execution_requests = ctx - .execution_block_generator - .read() - .get_execution_requests(&id); + let core = ctx.core_get_payload(id)?; + let fork = core.fork; // validate method called correctly according to shanghai fork time - if ctx - .execution_block_generator - .read() - .get_fork_at_timestamp(response.timestamp()) - == ForkName::Capella - && method == ENGINE_GET_PAYLOAD_V1 - { + if fork == ForkName::Capella && method == ENGINE_GET_PAYLOAD_V1 { return Err(( format!("{} called after Capella fork!", method), FORK_REQUEST_MISMATCH_ERROR_CODE, )); } // validate method called correctly according to cancun fork time - if ctx - .execution_block_generator - .read() - .get_fork_at_timestamp(response.timestamp()) - == ForkName::Deneb + if fork == ForkName::Deneb && (method == ENGINE_GET_PAYLOAD_V1 || method == ENGINE_GET_PAYLOAD_V2) { return Err(( @@ -316,11 +264,7 @@ pub async fn handle_rpc( )); } // validate method called correctly according to prague fork time - if ctx - .execution_block_generator - .read() - .get_fork_at_timestamp(response.timestamp()) - == ForkName::Electra + if fork == ForkName::Electra && (method == ENGINE_GET_PAYLOAD_V1 || method == ENGINE_GET_PAYLOAD_V2 || method == ENGINE_GET_PAYLOAD_V3) @@ -332,11 +276,7 @@ pub async fn handle_rpc( } // validate method called correctly according to osaka fork time - if ctx - .execution_block_generator - .read() - .get_fork_at_timestamp(response.timestamp()) - == ForkName::Fulu + if fork == ForkName::Fulu && (method == ENGINE_GET_PAYLOAD_V1 || method == ENGINE_GET_PAYLOAD_V2 || method == ENGINE_GET_PAYLOAD_V3 @@ -349,11 +289,7 @@ pub async fn handle_rpc( } // validate method called correctly according to amsterdam fork time - if ctx - .execution_block_generator - .read() - .get_fork_at_timestamp(response.timestamp()) - == ForkName::Gloas + if fork == ForkName::Gloas && (method == ENGINE_GET_PAYLOAD_V1 || method == ENGINE_GET_PAYLOAD_V2 || method == ENGINE_GET_PAYLOAD_V3 @@ -367,11 +303,7 @@ pub async fn handle_rpc( } // validate method called correctly according to heze fork time - if ctx - .execution_block_generator - .read() - .get_fork_at_timestamp(response.timestamp()) - == ForkName::Heze + if fork == ForkName::Heze && (method == ENGINE_GET_PAYLOAD_V1 || method == ENGINE_GET_PAYLOAD_V2 || method == ENGINE_GET_PAYLOAD_V3 @@ -386,144 +318,17 @@ pub async fn handle_rpc( match method { ENGINE_GET_PAYLOAD_V1 => Ok(serde_json::to_value( - JsonExecutionPayload::try_from(response).unwrap(), + JsonExecutionPayload::try_from(core.payload).unwrap(), ) .unwrap()), - ENGINE_GET_PAYLOAD_V2 => { - Ok(match JsonExecutionPayload::try_from(response).unwrap() { - JsonExecutionPayload::Bellatrix(execution_payload) => { - serde_json::to_value(JsonGetPayloadResponseBellatrix { - execution_payload, - block_value: Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI), - }) - .unwrap() - } - JsonExecutionPayload::Capella(execution_payload) => { - serde_json::to_value(JsonGetPayloadResponseCapella { - execution_payload, - block_value: Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI), - }) - .unwrap() - } - _ => unreachable!(), - }) - } - // From v3 onwards, we use the getPayload version only for the corresponding - // ExecutionPayload version. So we return an error if the ExecutionPayload version - // we get does not correspond to the getPayload version. - ENGINE_GET_PAYLOAD_V3 => { - Ok(match JsonExecutionPayload::try_from(response).unwrap() { - JsonExecutionPayload::Deneb(execution_payload) => { - serde_json::to_value(JsonGetPayloadResponseDeneb { - execution_payload, - block_value: Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI), - blobs_bundle: maybe_blobs - .ok_or(( - "No blobs returned despite V3 Payload".to_string(), - GENERIC_ERROR_CODE, - ))? - .into(), - should_override_builder: false, - }) - .unwrap() - } - _ => unreachable!(), - }) - } - ENGINE_GET_PAYLOAD_V4 => { - Ok(match JsonExecutionPayload::try_from(response).unwrap() { - JsonExecutionPayload::Electra(execution_payload) => { - serde_json::to_value(JsonGetPayloadResponseElectra { - execution_payload, - block_value: Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI), - blobs_bundle: maybe_blobs - .ok_or(( - "No blobs returned despite V4 Payload".to_string(), - GENERIC_ERROR_CODE, - ))? - .into(), - should_override_builder: false, - execution_requests: maybe_execution_requests - .clone() - .unwrap_or_else(|| { - types::ExecutionRequests::Electra(Default::default()) - }) - .into(), - }) - .unwrap() - } - _ => unreachable!(), - }) - } - ENGINE_GET_PAYLOAD_V5 => { - Ok(match JsonExecutionPayload::try_from(response).unwrap() { - JsonExecutionPayload::Fulu(execution_payload) => { - serde_json::to_value(JsonGetPayloadResponseFulu { - execution_payload, - block_value: Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI), - blobs_bundle: maybe_blobs - .ok_or(( - "No blobs returned despite V5 Payload".to_string(), - GENERIC_ERROR_CODE, - ))? - .into(), - should_override_builder: false, - execution_requests: maybe_execution_requests - .clone() - .unwrap_or_else(|| { - types::ExecutionRequests::Electra(Default::default()) - }) - .into(), - }) - .unwrap() - } - _ => unreachable!(), - }) - } - ENGINE_GET_PAYLOAD_V6 => { - Ok(match JsonExecutionPayload::try_from(response).unwrap() { - JsonExecutionPayload::Gloas(execution_payload) => { - serde_json::to_value(JsonGetPayloadResponseGloas { - execution_payload, - block_value: Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI), - blobs_bundle: maybe_blobs - .ok_or(( - "No blobs returned despite V6 Payload".to_string(), - GENERIC_ERROR_CODE, - ))? - .into(), - should_override_builder: false, - execution_requests: maybe_execution_requests - .unwrap_or_else(|| { - types::ExecutionRequests::Electra(Default::default()) - }) - .into(), - }) - .unwrap() - } - JsonExecutionPayload::Heze(execution_payload) => { - serde_json::to_value(JsonGetPayloadResponseHeze { - execution_payload, - block_value: Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI), - blobs_bundle: maybe_blobs - .ok_or(( - "No blobs returned despite V6 Payload".to_string(), - GENERIC_ERROR_CODE, - ))? - .into(), - should_override_builder: false, - execution_requests: maybe_execution_requests - .unwrap_or_else(|| { - types::ExecutionRequests::Electra(Default::default()) - }) - .into(), - }) - .unwrap() - } - _ => unreachable!(), - }) + // V2 onwards wraps the payload in a getPayload response; the version is validated + // against the fork above, so the shared conversion yields the matching variant. + _ => { + let response = + JsonGetPayloadResponse::try_from(core.into_get_payload_response()) + .map_err(|e| (format!("{e:?}"), GENERIC_ERROR_CODE))?; + Ok(serde_json::to_value(response).unwrap()) } - _ => unreachable!(), } } ENGINE_GET_BLOBS_V2 => { @@ -729,45 +534,10 @@ pub async fn handle_rpc( }; } - if let Some(hook_response) = ctx - .hook - .lock() - .on_forkchoice_updated(forkchoice_state.clone(), payload_attributes.clone()) - { - return Ok(serde_json::to_value(hook_response).unwrap()); - } - - let head_block_hash = forkchoice_state.head_block_hash; - - // Canned responses set by block hash take priority. - if let Some(status) = ctx.get_fcu_payload_status(&head_block_hash) { - return status - .map(|status| { - let response = JsonForkchoiceUpdatedV1Response { - payload_status: JsonPayloadStatusV1::from(status), - payload_id: None, - }; - serde_json::to_value(response).unwrap() - }) - .map_err(|message| (message, GENERIC_ERROR_CODE)); - } - - let mut response = ctx - .execution_block_generator - .write() - .forkchoice_updated( - forkchoice_state.into(), - payload_attributes.map(|json| json.into()), - ) - .map_err(|s| (s, GENERIC_ERROR_CODE))?; - - if let Some(mut status) = ctx.static_forkchoice_updated_response.lock().clone() { - if status.status == PayloadStatusV1Status::Valid { - status.latest_valid_hash = Some(head_block_hash) - } - - response.payload_status = status.into(); - } + let response = ctx.core_forkchoice_updated( + forkchoice_state.into(), + payload_attributes.map(|json| json.into()), + )?; Ok(serde_json::to_value(response).unwrap()) } diff --git a/beacon_node/execution_layer/src/test_utils/hook.rs b/beacon_node/execution_layer/src/test_utils/hook.rs index 4653811ac90..be3704d60d5 100644 --- a/beacon_node/execution_layer/src/test_utils/hook.rs +++ b/beacon_node/execution_layer/src/test_utils/hook.rs @@ -1,9 +1,8 @@ +use crate::engine_api::PayloadAttributes; +use crate::engines::ForkchoiceState; use crate::json_structures::*; -type ForkChoiceUpdatedHook = dyn Fn( - JsonForkchoiceStateV1, - Option, - ) -> Option +type ForkChoiceUpdatedHook = dyn Fn(ForkchoiceState, Option) -> Option + Send + Sync; @@ -15,8 +14,8 @@ pub struct Hook { impl Hook { pub fn on_forkchoice_updated( &self, - state: JsonForkchoiceStateV1, - payload_attributes: Option, + state: ForkchoiceState, + payload_attributes: Option, ) -> Option { (self.forkchoice_updated.as_ref()?)(state, payload_attributes) } diff --git a/beacon_node/execution_layer/src/test_utils/mock_engine_core.rs b/beacon_node/execution_layer/src/test_utils/mock_engine_core.rs new file mode 100644 index 00000000000..0b55fbe885b --- /dev/null +++ b/beacon_node/execution_layer/src/test_utils/mock_engine_core.rs @@ -0,0 +1,213 @@ +use crate::engine_api::{ + GetPayloadResponse, GetPayloadResponseBellatrix, GetPayloadResponseCapella, + GetPayloadResponseDeneb, GetPayloadResponseElectra, GetPayloadResponseFulu, + GetPayloadResponseGloas, GetPayloadResponseHeze, PayloadAttributes, PayloadId, + PayloadStatusV1Status, +}; +use crate::engines::ForkchoiceState; +use crate::json_structures::{JsonForkchoiceUpdatedV1Response, JsonPayloadStatusV1}; +use crate::test_utils::handle_rpc::{GENERIC_ERROR_CODE, UNKNOWN_PAYLOAD_ERROR_CODE}; +use crate::{BlobsBundle, PayloadStatusV1}; +use types::{ + EthSpec, ExecutionPayload, ExecutionRequests, ExecutionRequestsElectra, ExecutionRequestsGloas, + ForkName, Uint256, +}; + +use super::{Context, DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI}; + +pub(crate) struct CorePayload { + pub payload: ExecutionPayload, + pub blobs: Option>, + pub requests: Option>, + pub fork: ForkName, +} + +impl CorePayload { + pub(crate) fn into_get_payload_response(self) -> GetPayloadResponse { + let CorePayload { + payload, + blobs, + requests, + fork: _, + } = self; + let block_value = Uint256::from(DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI); + let blobs_bundle = blobs.unwrap_or_default(); + + match payload { + ExecutionPayload::Bellatrix(execution_payload) => { + GetPayloadResponse::Bellatrix(GetPayloadResponseBellatrix { + execution_payload, + block_value, + }) + } + ExecutionPayload::Capella(execution_payload) => { + GetPayloadResponse::Capella(GetPayloadResponseCapella { + execution_payload, + block_value, + }) + } + ExecutionPayload::Deneb(execution_payload) => { + GetPayloadResponse::Deneb(GetPayloadResponseDeneb { + execution_payload, + block_value, + blobs_bundle, + should_override_builder: false, + }) + } + ExecutionPayload::Electra(execution_payload) => { + GetPayloadResponse::Electra(GetPayloadResponseElectra { + execution_payload, + block_value, + blobs_bundle, + should_override_builder: false, + requests: electra_requests(requests), + }) + } + ExecutionPayload::Fulu(execution_payload) => { + GetPayloadResponse::Fulu(GetPayloadResponseFulu { + execution_payload, + block_value, + blobs_bundle, + should_override_builder: false, + requests: electra_requests(requests), + }) + } + ExecutionPayload::Gloas(execution_payload) => { + GetPayloadResponse::Gloas(GetPayloadResponseGloas { + execution_payload, + block_value, + blobs_bundle, + should_override_builder: false, + requests: gloas_requests(requests), + }) + } + ExecutionPayload::Heze(execution_payload) => { + GetPayloadResponse::Heze(GetPayloadResponseHeze { + execution_payload, + block_value, + blobs_bundle, + should_override_builder: false, + requests: gloas_requests(requests), + }) + } + } + } +} + +fn electra_requests( + requests: Option>, +) -> ExecutionRequestsElectra { + match requests { + Some(ExecutionRequests::Electra(inner)) => inner, + _ => ExecutionRequestsElectra::default(), + } +} + +fn gloas_requests(requests: Option>) -> ExecutionRequestsGloas { + match requests { + Some(ExecutionRequests::Gloas(inner)) => inner, + _ => ExecutionRequestsGloas::default(), + } +} + +impl Context { + pub(crate) fn core_new_payload( + &self, + payload: ExecutionPayload, + ) -> Result { + // Canned responses set by block hash take priority. + if let Some(status) = self.get_new_payload_status(&payload.block_hash()) { + return status.map_err(|message| (message, GENERIC_ERROR_CODE)); + } + + let (static_response, should_import) = + if let Some(mut response) = self.static_new_payload_response.lock().clone() { + if response.status.status == PayloadStatusV1Status::Valid { + response.status.latest_valid_hash = Some(payload.block_hash()) + } + + (Some(response.status), response.should_import) + } else { + (None, true) + }; + + let dynamic_response = if should_import { + Some(self.execution_block_generator.write().new_payload(payload)) + } else { + None + }; + + Ok(static_response.or(dynamic_response).unwrap()) + } + + pub(crate) fn core_forkchoice_updated( + &self, + state: ForkchoiceState, + attrs: Option, + ) -> Result { + *self.previous_forkchoice_request.lock() = Some((state, attrs.clone())); + + if let Some(hook_response) = self.hook.lock().on_forkchoice_updated(state, attrs.clone()) { + return Ok(hook_response); + } + + let head_block_hash = state.head_block_hash; + + // Canned responses set by block hash take priority. + if let Some(status) = self.get_fcu_payload_status(&head_block_hash) { + return status + .map(|status| JsonForkchoiceUpdatedV1Response { + payload_status: JsonPayloadStatusV1::from(status), + payload_id: None, + }) + .map_err(|message| (message, GENERIC_ERROR_CODE)); + } + + let mut response = self + .execution_block_generator + .write() + .forkchoice_updated(state, attrs) + .map_err(|s| (s, GENERIC_ERROR_CODE))?; + + if let Some(mut status) = self.static_forkchoice_updated_response.lock().clone() { + if status.status == PayloadStatusV1Status::Valid { + status.latest_valid_hash = Some(head_block_hash); + } + + response.payload_status = status.into(); + } + + Ok(response) + } + + pub(crate) fn core_get_payload(&self, id: PayloadId) -> Result, (String, i64)> { + let payload = self + .execution_block_generator + .write() + .get_payload(&id) + .ok_or_else(|| { + ( + format!("no payload for id {:?}", id), + UNKNOWN_PAYLOAD_ERROR_CODE, + ) + })?; + + let blobs = self.execution_block_generator.write().get_blobs_bundle(&id); + let requests = self + .execution_block_generator + .read() + .get_execution_requests(&id); + + let fork = self + .execution_block_generator + .read() + .get_fork_at_timestamp(payload.timestamp()); + + Ok(CorePayload { + payload, + blobs, + requests, + fork, + }) + } +} diff --git a/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs b/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs index de0e69016e5..19c5eec18de 100644 --- a/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs +++ b/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs @@ -1,4 +1,6 @@ -use crate::{test_utils::DEFAULT_JWT_SECRET, test_utils::MockServer, *}; +use crate::{ + test_utils::DEFAULT_JWT_SECRET, test_utils::MockExecutionConfig, test_utils::MockServer, *, +}; use alloy_primitives::B256 as H256; use fixed_bytes::FixedBytesExtended; use kzg::Kzg; @@ -12,6 +14,30 @@ pub struct MockExecutionLayer { pub spec: Arc, } +/// Env var that selects the engine transport for every test-infra `MockExecutionLayer`. +/// +/// Unset or `JSON` runs JSON-RPC (the default); `SSZ` runs the REST-SSZ mock engine. Any other +/// value panics rather than silently falling back. Matching is case-insensitive. Because all +/// mocks — including the `beacon_chain` harness's `mock_execution_layer_from_parts` — funnel +/// through `MockExecutionLayer::new`, setting this flips the transport across the whole test tree +/// with no per-crate Cargo plumbing, e.g. `LH_ENGINE_TRANSPORT=SSZ cargo nextest run`. +pub const ENGINE_TRANSPORT_ENV_VAR: &str = "LH_ENGINE_TRANSPORT"; + +/// `true` when `LH_ENGINE_TRANSPORT` selects the REST-SSZ mock transport (`SSZ`); `false` when +/// unset or `JSON`. Panics on any other value so a typo can't quietly run the wrong transport. +pub fn mock_rest_ssz_enabled() -> bool { + match std::env::var(ENGINE_TRANSPORT_ENV_VAR) { + Ok(value) => match value.trim().to_ascii_uppercase().as_str() { + "SSZ" => true, + "" | "JSON" => false, + other => { + panic!("invalid {ENGINE_TRANSPORT_ENV_VAR}={other:?}; expected \"JSON\" or \"SSZ\"") + } + }, + Err(_) => false, + } +} + impl MockExecutionLayer { pub fn default_params(executor: TaskExecutor) -> Self { let mut spec = MainnetEthSpec::default_spec(); @@ -31,6 +57,46 @@ impl MockExecutionLayer { ) } + pub fn fulu_params(executor: TaskExecutor) -> Self { + let mut spec = MainnetEthSpec::default_spec(); + spec.terminal_block_hash = ExecutionBlockHash::zero(); + spec.terminal_block_hash_activation_epoch = Epoch::new(0); + Self::new( + executor, + Some(0), + Some(0), + Some(0), + Some(0), + None, + None, + Some(JwtKey::from_slice(&DEFAULT_JWT_SECRET).unwrap()), + Arc::new(spec), + None, + ) + } + + /// Build a mock whose client advertises REST-SSZ but whose server does not serve it. The + /// capabilities probe then 404s over REST, forcing the transport to fall back to JSON-RPC. + pub fn rest_client_without_rest_server(executor: TaskExecutor) -> Self { + let mut spec = MainnetEthSpec::default_spec(); + spec.terminal_block_hash = ExecutionBlockHash::zero(); + spec.terminal_block_hash_activation_epoch = Epoch::new(0); + Self::new_with_transport( + executor, + None, + None, + None, + None, + None, + None, + Some(JwtKey::from_slice(&DEFAULT_JWT_SECRET).unwrap()), + Arc::new(spec), + None, + true, + false, + ) + } + #[allow(clippy::too_many_arguments)] pub fn new( executor: TaskExecutor, @@ -44,18 +110,54 @@ impl MockExecutionLayer { spec: Arc, kzg: Option>, ) -> Self { - let handle = executor.handle().unwrap(); - - let jwt_key = jwt_key.unwrap_or_else(JwtKey::random); - let server = MockServer::new( - &handle, - jwt_key, + let rest_ssz = mock_rest_ssz_enabled(); + Self::new_with_transport( + executor, shanghai_time, cancun_time, prague_time, osaka_time, amsterdam_time, heze_time, + jwt_key, + spec, + kzg, + rest_ssz, + rest_ssz, + ) + } + + #[allow(clippy::too_many_arguments)] + fn new_with_transport( + executor: TaskExecutor, + shanghai_time: Option, + cancun_time: Option, + prague_time: Option, + osaka_time: Option, + amsterdam_time: Option, + heze_time: Option, + jwt_key: Option, + spec: Arc, + kzg: Option>, + client_rest_ssz: bool, + server_rest_ssz: bool, + ) -> Self { + let handle = executor.handle().unwrap(); + + let jwt_key = jwt_key.unwrap_or_else(JwtKey::random); + let server = MockServer::new_with_config( + &handle, + MockExecutionConfig { + jwt_key, + shanghai_time, + cancun_time, + prague_time, + osaka_time, + amsterdam_time, + heze_time, + serve_rest_ssz: server_rest_ssz, + ..Default::default() + }, kzg, ); @@ -69,6 +171,7 @@ impl MockExecutionLayer { execution_endpoint: Some(url), secret_file: Some(path), suggested_fee_recipient: Some(Address::repeat_byte(42)), + engine_api_rest_ssz: client_rest_ssz, ..Default::default() }; let el = ExecutionLayer::from_config(config, executor.clone()).unwrap(); @@ -81,6 +184,19 @@ impl MockExecutionLayer { } } + /// Resolve the engine transport with an **awaited** upcheck, then assert it matches the mode + /// selected by `LH_ENGINE_TRANSPORT` (`Rest` when `SSZ`, `JsonRpcOnly` otherwise). + pub async fn resolve_and_assert_transport(self) -> Self { + self.el.upcheck().await; + let expected = if mock_rest_ssz_enabled() { + Transport::Rest + } else { + Transport::JsonRpcOnly + }; + assert_eq!(self.el.resolved_transport(), Some(expected)); + self + } + pub async fn produce_valid_execution_payload_on_head(self) -> Self { let latest_execution_block = { let block_gen = self.server.execution_block_generator(); @@ -267,6 +383,106 @@ impl MockExecutionLayer { self } + pub async fn reconcile_unknown_payload_on_head(self) -> Self { + let latest_execution_block = { + let block_gen = self.server.execution_block_generator(); + block_gen.latest_block().unwrap() + }; + + let parent_hash = latest_execution_block.block_hash(); + let parent_gas_limit = latest_execution_block.gas_limit(); + let block_number = latest_execution_block.block_number() + 1; + let timestamp = block_number; + let prev_randao = Hash256::from_low_u64_be(block_number); + let head_block_root = Hash256::repeat_byte(42); + let slot = Slot::new(0); + let validator_index = 0; + + let current_fork = self + .server + .execution_block_generator() + .get_fork_at_timestamp(timestamp); + + let forkchoice_update_params = ForkchoiceUpdateParameters { + head_root: head_block_root, + head_hash: Some(parent_hash), + justified_hash: None, + finalized_hash: None, + }; + let suggested_fee_recipient = self.el.get_suggested_fee_recipient(validator_index).await; + let payload_attributes = PayloadAttributes::new( + timestamp, + prev_randao, + suggested_fee_recipient, + current_fork.capella_enabled().then(Vec::new), + current_fork.deneb_enabled().then(Hash256::zero), + None, + None, + None, + ); + + let build_params = || PayloadParameters { + parent_hash, + parent_gas_limit: Some(parent_gas_limit), + proposer_gas_limit: None, + payload_attributes: &payload_attributes, + forkchoice_update_params: &forkchoice_update_params, + current_fork, + }; + let builder_params = || BuilderParams { + pubkey: PublicKeyBytes::empty(), + slot, + chain_health: ChainHealth::Healthy, + }; + + self.el + .get_payload( + build_params(), + builder_params(), + &self.spec, + None, + BlockProductionVersion::FullV2, + ) + .await + .unwrap(); + + let expired_before = crate::metrics::get_int_counter( + &crate::metrics::EXECUTION_LAYER_PRE_PREPARED_PAYLOAD_ID, + &[crate::metrics::EXPIRED], + ) + .map(|counter| counter.get()) + .unwrap_or(0); + + // Make the cached id unknown to the EL so the next `get_payload` reconciles. + self.server.expire_all_payload_ids(); + + // Same `(parent_hash, attrs)`: cache HIT the stale id -> unknown-payload -> re-fcU -> retry once. + self.el + .get_payload( + build_params(), + builder_params(), + &self.spec, + None, + BlockProductionVersion::FullV2, + ) + .await + .unwrap(); + + let expired_after = crate::metrics::get_int_counter( + &crate::metrics::EXECUTION_LAYER_PRE_PREPARED_PAYLOAD_ID, + &[crate::metrics::EXPIRED], + ) + .map(|counter| counter.get()) + .unwrap_or(0); + assert_eq!( + expired_after, + expired_before + 1, + "reconcile did not fire: cached id was not observed as unknown-payload" + ); + + self + } + #[allow(clippy::too_many_arguments)] pub async fn assert_valid_execution_payload_on_head>( &self, diff --git a/beacon_node/execution_layer/src/test_utils/mod.rs b/beacon_node/execution_layer/src/test_utils/mod.rs index 1b139620281..de2e9a80be6 100644 --- a/beacon_node/execution_layer/src/test_utils/mod.rs +++ b/beacon_node/execution_layer/src/test_utils/mod.rs @@ -25,7 +25,7 @@ use tracing::info; use types::{EthSpec, ExecutionBlockHash, Uint256}; use warp::{Filter, Rejection, http::StatusCode}; -use crate::{EngineCapabilities, JsonRpcCapabilities}; +use crate::{EngineCapabilities, ForkchoiceState, JsonRpcCapabilities, PayloadAttributes}; pub use execution_block_generator::DEFAULT_GAS_LIMIT; pub use execution_block_generator::{ Block, ExecutionBlockGenerator, generate_blobs, generate_genesis_block, @@ -34,6 +34,7 @@ pub use execution_block_generator::{ pub use hook::Hook; pub use mock_builder::{MockBuilder, Operation, mock_builder_extra_data}; pub use mock_execution_layer::MockExecutionLayer; +pub use mock_execution_layer::mock_rest_ssz_enabled; pub const DEFAULT_JWT_SECRET: [u8; 32] = [42; 32]; pub const DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI: u128 = 10_000_000_000_000_000; @@ -79,9 +80,11 @@ pub static DEFAULT_CLIENT_VERSION: LazyLock = }); mod execution_block_generator; +mod handle_rest; mod handle_rpc; mod hook; mod mock_builder; +mod mock_engine_core; mod mock_execution_layer; /// Configuration for the MockExecutionLayer. @@ -173,6 +176,7 @@ impl MockServer { serve_rest_ssz, execution_block_generator: RwLock::new(execution_block_generator), previous_request: <_>::default(), + previous_forkchoice_request: <_>::default(), preloaded_responses, static_new_payload_response: <_>::default(), static_forkchoice_updated_response: <_>::default(), @@ -261,6 +265,13 @@ impl MockServer { self.ctx.execution_block_generator.write() } + pub fn expire_all_payload_ids(&self) { + self.ctx + .execution_block_generator + .write() + .expire_all_payload_ids(); + } + pub fn url(&self) -> String { format!( "http://{}:{}", @@ -292,6 +303,10 @@ impl MockServer { self.ctx.previous_request.lock().take() } + pub fn take_previous_forkchoice_request(&self) -> Option { + self.ctx.previous_forkchoice_request.lock().take() + } + pub fn set_new_payload_response(&self, response: StaticNewPayloadResponse) { *self.ctx.static_new_payload_response.lock() = Some(response) } @@ -564,6 +579,9 @@ struct AuthError(String); impl warp::reject::Reject for AuthError {} +/// A captured `forkchoice_updated` request: the state plus any payload attributes. +pub type CapturedForkchoiceRequest = (ForkchoiceState, Option); + /// A wrapper around all the items required to spawn the HTTP server. /// /// The server will gracefully handle the case where any fields are `None`. @@ -577,6 +595,7 @@ pub struct Context { pub execution_block_generator: RwLock>, pub preloaded_responses: Arc>>, pub previous_request: Arc>>, + pub previous_forkchoice_request: Arc>>, pub static_new_payload_response: Arc>>, pub static_forkchoice_updated_response: Arc>>, pub hook: Arc>, @@ -689,7 +708,9 @@ async fn handle_rejection(err: Rejection) -> Result( method: &'static str, @@ -717,16 +738,20 @@ async fn handle_rest_capture( *ctx.last_rest_request.write() = Some(RestCapture { method: method.to_string(), path, - eth_execution_version, + eth_execution_version: eth_execution_version.clone(), client_version, content_type, - body, + body: body.clone(), }); - Ok(warp::http::Response::builder() - .status(204) - .body(Bytes::new()) - .unwrap()) + Ok(handle_rest::handle_rest( + method, + full_path, + query.as_deref(), + eth_execution_version.as_deref(), + &body, + &ctx, + )) } /// Creates a server that will serve requests using information from `ctx`. diff --git a/beacon_node/http_api/tests/gloas_reorg_tests.rs b/beacon_node/http_api/tests/gloas_reorg_tests.rs index f31746ddf70..36dd23ae73e 100644 --- a/beacon_node/http_api/tests/gloas_reorg_tests.rs +++ b/beacon_node/http_api/tests/gloas_reorg_tests.rs @@ -514,8 +514,6 @@ pub async fn proposer_boost_re_org_test( .lock() .set_forkchoice_updated_hook(Box::new(move |state, payload_attributes| { let received_at = chain_inner.slot_clock.now_duration().unwrap(); - let state = ForkchoiceState::from(state); - let payload_attributes = payload_attributes.map(Into::into); let update = ForkChoiceUpdateMetadata { received_at, state, diff --git a/beacon_node/http_api/tests/interactive_tests.rs b/beacon_node/http_api/tests/interactive_tests.rs index 71aef5760d1..8ccd53bc5f7 100644 --- a/beacon_node/http_api/tests/interactive_tests.rs +++ b/beacon_node/http_api/tests/interactive_tests.rs @@ -464,8 +464,6 @@ pub async fn proposer_boost_re_org_test( .lock() .set_forkchoice_updated_hook(Box::new(move |state, payload_attributes| { let received_at = chain_inner.slot_clock.now_duration().unwrap(); - let state = ForkchoiceState::from(state); - let payload_attributes = payload_attributes.map(Into::into); let update = ForkChoiceUpdateMetadata { received_at, state,