Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 124 additions & 1 deletion crates/common/src/beacon/beacon_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ use std::{sync::Arc, task::Poll, time::Duration};

use ::ssz::Encode;
use alloy_primitives::B256;
use helix_types::{ForkName, LhConfig, VersionedSignedProposal, spec_from_config};
use helix_types::{
ForkName, LhConfig, SignedExecutionPayloadEnvelope, VersionedSignedProposal, spec_from_config,
};
use http::{Request, header::CONTENT_TYPE};
use http_body_util::Full;
use hyper::body::Bytes;
Expand All @@ -20,6 +22,8 @@ use crate::{
};

const CONSENSUS_VERSION_HEADER: &str = "eth-consensus-version";
// Always "false": helix always has blobs cached from the builder's own submission.
const BLOB_DATA_INCLUDED_HEADER: &str = "eth-blob-data-included";
const PUBLISH_BLOCK_TIMEOUT: Duration = Duration::from_secs(4);
const GET_TIMEOUT: Duration = Duration::from_secs(5);

Expand Down Expand Up @@ -112,6 +116,48 @@ impl BeaconClient {
}
}

/// Publishes a signed execution payload envelope SSZ-encoded, so a connected beacon node
/// broadcasts it to the `execution_payload` gossip topic on helix's behalf.
/// <https://github.com/ethereum/beacon-APIs/blob/master/apis/beacon/execution_payload/envelope_post.yaml>
pub async fn publish_execution_payload_envelope(
&self,
envelope: Arc<SignedExecutionPayloadEnvelope>,
fork: ForkName,
) -> Result<u16, BeaconClientError> {
let target = self.config.url.join("eth/v1/beacon/execution_payload_envelopes")?;
let body_bytes = Bytes::from(envelope.as_ssz_bytes());
let req = Request::builder()
.method("POST")
.uri(target.as_str())
.header(CONSENSUS_VERSION_HEADER, fork.to_string())
.header(BLOB_DATA_INCLUDED_HEADER, "false")
.header(CONTENT_TYPE, "application/octet-stream")
.body(Full::new(body_bytes))?;
let mut pending = self.http.send(&target, req)?.with_timeout(PUBLISH_BLOCK_TIMEOUT);

let (status, body) = loop {
match pending.poll_bytes() {
Poll::Pending => {}
Poll::Ready(Ok(r)) => break r,
Poll::Ready(Err(e)) => return Err(e.into()),
}
tokio::task::yield_now().await;
};

match status {
200 => Ok(200),
202 => {
let body_str = String::from_utf8_lossy(&body);
warn!("Envelope broadcast but not integrated: {body_str}");
Ok(202)
}
_ => {
let api_err: ApiError = serde_json::from_slice(&body)?;
Err(BeaconClientError::Api(api_err))
}
}
}

pub async fn get_chain_info(&self) -> Result<ChainInfo, BeaconClientError> {
let spec: BeaconResponse<LhConfig> = self.get("eth/v1/config/spec").await?;
let spec = spec_from_config(spec.data);
Expand All @@ -130,3 +176,80 @@ impl BeaconClient {
Ok(chain_info)
}
}

#[cfg(test)]
mod tests {
use helix_types::{BlsSignature, ExecutionPayloadEnvelope};
use httpmock::{Method::POST, MockServer};
use reqwest::Url;

use super::*;

fn test_client(url: Url) -> BeaconClient {
crate::utils::install_default_crypto_provider();
BeaconClient::new(BeaconClientConfig { url })
}

fn empty_envelope() -> Arc<SignedExecutionPayloadEnvelope> {
Arc::new(SignedExecutionPayloadEnvelope {
message: ExecutionPayloadEnvelope::empty(),
signature: BlsSignature::empty(),
})
}

#[tokio::test]
async fn publish_execution_payload_envelope_sends_ssz_with_fork_and_blob_headers() {
let server = MockServer::start();
let mock = server.mock(|when, then| {
when.method(POST)
.path("/eth/v1/beacon/execution_payload_envelopes")
.header("eth-consensus-version", "gloas")
.header("eth-blob-data-included", "false")
.header("content-type", "application/octet-stream");
then.status(200);
});

let client = test_client(Url::parse(&server.url("/")).unwrap());
let result =
client.publish_execution_payload_envelope(empty_envelope(), ForkName::Gloas).await;

mock.assert();
assert_eq!(result.unwrap(), 200);
}

#[tokio::test]
async fn publish_execution_payload_envelope_202_is_ok() {
let server = MockServer::start();
server.mock(|when, then| {
when.method(POST).path("/eth/v1/beacon/execution_payload_envelopes");
then.status(202).body("envelope failed integration but was broadcast");
});

let client = test_client(Url::parse(&server.url("/")).unwrap());
let result =
client.publish_execution_payload_envelope(empty_envelope(), ForkName::Gloas).await;

assert_eq!(result.unwrap(), 202);
}

#[tokio::test]
async fn publish_execution_payload_envelope_error_response_parses_api_error() {
let server = MockServer::start();
server.mock(|when, then| {
when.method(POST).path("/eth/v1/beacon/execution_payload_envelopes");
then.status(400).json_body(serde_json::json!({
"code": 400,
"message": "Invalid signed execution payload envelope"
}));
});

let client = test_client(Url::parse(&server.url("/")).unwrap());
let result =
client.publish_execution_payload_envelope(empty_envelope(), ForkName::Gloas).await;

match result {
Err(BeaconClientError::Api(ApiError::ErrorMessage { code: 400, .. })) => {}
other => panic!("expected a 400 ApiError, got {other:?}"),
}
}
}
88 changes: 87 additions & 1 deletion crates/common/src/beacon/multi_beacon_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ use std::sync::{
};

use futures::future::join_all;
use helix_types::{ForkName, VersionedSignedProposal};
use helix_types::{ForkName, SignedExecutionPayloadEnvelope, VersionedSignedProposal};

use crate::{
beacon::{beacon_client::BeaconClient, error::BeaconClientError, types::BroadcastValidation},
Expand Down Expand Up @@ -83,4 +83,90 @@ impl MultiBeaconClient {

Err(last_error.unwrap_or(BeaconClientError::BeaconNodeUnavailable))
}

/// Publishes the signed execution payload envelope to all beacon clients; returns on first
/// success. Unlike `publish_block`, fans out via plain concurrent futures, not
/// `spawn_tracked!`.
pub async fn publish_execution_payload_envelope(
&self,
envelope: Arc<SignedExecutionPayloadEnvelope>,
fork: ForkName,
) -> Result<(), BeaconClientError> {
let futures = self
.beacon_clients
.iter()
.map(|client| client.publish_execution_payload_envelope(envelope.clone(), fork));

let mut last_error: Option<BeaconClientError> = None;
for res in join_all(futures).await {
match res {
Ok(_) => return Ok(()),
Err(err) => last_error = Some(err),
}
}

Err(last_error.unwrap_or(BeaconClientError::BeaconNodeUnavailable))
}
}

#[cfg(test)]
mod tests {
use helix_types::{BlsSignature, ExecutionPayloadEnvelope};
use httpmock::{Method::POST, MockServer};
use reqwest::Url;

use super::*;
use crate::BeaconClientConfig;

fn envelope() -> Arc<SignedExecutionPayloadEnvelope> {
Arc::new(SignedExecutionPayloadEnvelope {
message: ExecutionPayloadEnvelope::empty(),
signature: BlsSignature::empty(),
})
}

fn client_for(server: &MockServer) -> Arc<BeaconClient> {
let url = Url::parse(&server.url("/")).unwrap();
Arc::new(BeaconClient::new(BeaconClientConfig { url }))
}

#[tokio::test]
async fn publish_execution_payload_envelope_returns_ok_on_first_success() {
crate::utils::install_default_crypto_provider();
let failing = MockServer::start();
failing.mock(|when, then| {
when.method(POST).path("/eth/v1/beacon/execution_payload_envelopes");
then.status(500);
});
let succeeding = MockServer::start();
succeeding.mock(|when, then| {
when.method(POST).path("/eth/v1/beacon/execution_payload_envelopes");
then.status(200);
});

let multi = MultiBeaconClient::new(vec![client_for(&failing), client_for(&succeeding)]);
let result = multi.publish_execution_payload_envelope(envelope(), ForkName::Gloas).await;

assert!(result.is_ok(), "expected Ok, got {result:?}");
}

#[tokio::test]
async fn publish_execution_payload_envelope_returns_err_when_all_clients_fail() {
crate::utils::install_default_crypto_provider();
let a = MockServer::start();
a.mock(|when, then| {
when.method(POST).path("/eth/v1/beacon/execution_payload_envelopes");
then.status(500);
});
let b = MockServer::start();
b.mock(|when, then| {
when.method(POST).path("/eth/v1/beacon/execution_payload_envelopes");
then.status(500);
});

let multi = MultiBeaconClient::new(vec![client_for(&a), client_for(&b)]);
let result = multi.publish_execution_payload_envelope(envelope(), ForkName::Gloas).await;

assert!(result.is_err(), "expected Err, got {result:?}");
}
}
5 changes: 5 additions & 0 deletions crates/common/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,10 @@ pub struct RelayConfig {
pub enable_flux_profiler: bool,
#[serde(default)]
pub operator_config: Option<OperatorConfig>,
/// This relay's on-chain Gloas (ePBS) builder_index. Placeholder until helix has a real
/// on-chain builder registration; signs under the relay's own key in the meantime.
#[serde(default)]
pub gloas_builder_index: u64,
}

#[derive(Serialize, Deserialize, Clone)]
Expand Down Expand Up @@ -131,6 +135,7 @@ impl RelayConfig {
clickhouse: None,
enable_flux_profiler: false,
operator_config: None,
gloas_builder_index: 0,
}
}
}
Expand Down
19 changes: 18 additions & 1 deletion crates/relay/src/api/proposer/error.rs
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use alloy_primitives::B256;
use axum::{
self,
response::{IntoResponse, Response},
Expand Down Expand Up @@ -136,6 +137,19 @@ pub enum ProposerApiError {

#[error("invalid request: Date-Milliseconds and X-Timeout-Ms headers are required")]
MissingTimingHeaders,

#[error("no held execution payload for bid block hash {0:?}")]
NoHeldPayloadForBlock(B256),

#[error(
"held payload block hash {held:?} does not match the bid's committed block hash {bid:?}"
)]
HeldPayloadBlockHashMismatch { held: B256, bid: B256 },

#[error(
"bid builder_index {bid} does not match this relay's configured builder_index {configured}"
)]
BuilderIndexMismatch { bid: u64, configured: u64 },
}

impl From<DecodeError> for ProposerApiError {
Expand Down Expand Up @@ -181,7 +195,10 @@ impl IntoResponse for ProposerApiError {
ProposerApiError::GetPayloadAlreadyReceived |
ProposerApiError::RequestForPastSlot { .. } |
ProposerApiError::RequestAuthSlotMismatch { .. } |
ProposerApiError::MissingTimingHeaders => StatusCode::BAD_REQUEST,
ProposerApiError::MissingTimingHeaders |
ProposerApiError::NoHeldPayloadForBlock(_) |
ProposerApiError::HeldPayloadBlockHashMismatch { .. } |
ProposerApiError::BuilderIndexMismatch { .. } => StatusCode::BAD_REQUEST,

// All authentication failures, kept indistinguishable by status
ProposerApiError::InvalidApiKey |
Expand Down
10 changes: 10 additions & 0 deletions crates/relay/src/api/proposer/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ use helix_common::{
use helix_database::handle::DbHandle;
use helix_operator::OperatorPubSub;
use hyper::StatusCode;
pub use submit_signed_beacon_block::{GloasBuilderIdentity, GloasPayloadStore, NoHeldPayloads};

use crate::{
api::{Api, router::Terminating},
Expand All @@ -44,6 +45,8 @@ pub struct ProposerApi<A: Api> {
pub auctioneer_handle: AuctioneerHandle,
pub reg_handle: RegWorkerHandle,
pub operator_api: Option<Arc<OperatorPubSub>>,
pub gloas_builder_identity: Arc<GloasBuilderIdentity>,
pub gloas_payload_store: Arc<dyn GloasPayloadStore>,
}

impl<A: Api> ProposerApi<A> {
Expand All @@ -62,7 +65,12 @@ impl<A: Api> ProposerApi<A> {
reg_handle: RegWorkerHandle,
alert_manager: Arc<AlertManager>,
operator_api: Option<Arc<OperatorPubSub>>,
gloas_payload_store: Arc<dyn GloasPayloadStore>,
) -> Self {
let gloas_builder_identity = Arc::new(GloasBuilderIdentity {
builder_index: relay_config.gloas_builder_index,
keypair: signing_context.keypair.clone(),
});
Self {
local_cache,
db,
Expand All @@ -78,6 +86,8 @@ impl<A: Api> ProposerApi<A> {
auctioneer_handle,
reg_handle,
operator_api,
gloas_builder_identity,
gloas_payload_store,
}
}
}
Expand Down
Loading
Loading