feat: add relay monitors (#73)

* add relay monitor support

* use `Url` instead of strings

Author: ltitanb

Cargo.toml

@@ -71,7 +71,7 @@ clap = { version = "4.5.4", features = ["derive", "env"] }
 thiserror = "1.0.61"
 color-eyre = "0.6.3"
 eyre = "0.6.12"
-url = "2.5.0"
+url = { version = "2.5.0", features = ["serde"] }
 uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] }
 typenum = "1.17.0"
 rand = "0.8.5"

config.example.toml

@@ -34,6 +34,9 @@ skip_sigverify = false
 # Minimum bid in ETH that will be accepted from `get_header`
 # OPTIONAL, DEFAULT: 0.0
 min_bid_eth = 0.0
+# List of URLs of relay monitors to send registrations to
+# OPTIONAL
+relay_monitors = []
 # How late in milliseconds in the slot is "late". This impacts the `get_header` requests, by shortening timeouts for `get_header` calls to 
 # relays and make sure a header is returned within this deadline. If the request from the CL comes later in the slot, then fetching headers is skipped
 # to force local building and miniminzing the risk of missed slots. See also the timing games section below

crates/common/src/config/mod.rs

@@ -34,11 +34,21 @@ pub struct CommitBoostConfig {
 }
 
 impl CommitBoostConfig {
+    /// Validate config
+    pub fn validate(&self) -> Result<()> {
+        self.pbs.pbs_config.validate()?;
+        Ok(())
+    }
+
     pub fn from_file(path: &str) -> Result<Self> {
-        load_from_file(path)
+        let config: Self = load_from_file(path)?;
+        config.validate()?;
+        Ok(config)
     }
 
     pub fn from_env_path() -> Result<Self> {
-        load_file_from_env(CB_CONFIG_ENV)
+        let config: Self = load_file_from_env(CB_CONFIG_ENV)?;
+        config.validate()?;
+        Ok(config)
     }
 }

crates/common/src/config/pbs.rs

@@ -5,6 +5,7 @@ use std::{collections::HashMap, sync::Arc};
 use alloy::primitives::U256;
 use eyre::Result;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use url::Url;
 
 use super::{constants::PBS_DEFAULT_IMAGE, CommitBoostConfig};
 use crate::{
@@ -15,7 +16,7 @@ use crate::{
     utils::{as_eth_str, default_bool, default_u256, default_u64},
 };
 
-#[derive(Debug, Clone, Default, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct RelayConfig {
     /// Relay ID, if missing will default to the URL hostname from the entry
     pub id: Option<String>,
@@ -33,7 +34,7 @@ pub struct RelayConfig {
     pub frequency_get_header_ms: Option<u64>,
 }
 
-#[derive(Debug, Clone, Default, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct PbsConfig {
     /// Port to receive BuilderAPI calls from beacon node
     pub port: u16,
@@ -55,13 +56,23 @@ pub struct PbsConfig {
     /// Minimum bid that will be accepted from get_header
     #[serde(rename = "min_bid_eth", with = "as_eth_str", default = "default_u256")]
     pub min_bid_wei: U256,
+    /// List of relay monitor urls in the form of scheme://host
+    #[serde(default)]
+    pub relay_monitors: Vec<Url>,
     /// How late in the slot we consider to be "late"
     #[serde(default = "default_u64::<LATE_IN_SLOT_TIME_MS>")]
     pub late_in_slot_time_ms: u64,
 }
 
+impl PbsConfig {
+    /// Validate PBS config parameters
+    pub fn validate(&self) -> Result<()> {
+        Ok(())
+    }
+}
+
 /// Static pbs config from config file
-#[derive(Debug, Default, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize)]
 pub struct StaticPbsConfig {
     /// Docker image of the module
     #[serde(default = "default_pbs")]
@@ -96,6 +107,7 @@ fn default_pbs() -> String {
 /// Loads the default pbs config, i.e. with no signer client or custom data
 pub fn load_pbs_config() -> Result<PbsModuleConfig> {
     let config = CommitBoostConfig::from_env_path()?;
+
     let relay_clients =
         config.relays.into_iter().map(RelayClient::new).collect::<Result<Vec<_>>>()?;
     let maybe_publiher = BuilderEventPublisher::new_from_env();
@@ -128,6 +140,8 @@ pub fn load_pbs_custom_config<T: DeserializeOwned>() -> Result<(PbsModuleConfig,
 
     // load module config including the extra data (if any)
     let cb_config: StubConfig<T> = load_file_from_env(CB_CONFIG_ENV)?;
+    cb_config.pbs.static_config.pbs_config.validate()?;
+
     let relay_clients =
         cb_config.relays.into_iter().map(RelayClient::new).collect::<Result<Vec<_>>>()?;
     let maybe_publiher = BuilderEventPublisher::new_from_env();

crates/common/src/pbs/constants.rs

@@ -1,4 +1,4 @@
-pub const BULDER_API_PATH: &str = "/eth/v1/builder";
+pub const BUILDER_API_PATH: &str = "/eth/v1/builder";
 
 pub const GET_HEADER_PATH: &str = "/header/:slot/:parent_hash/:pubkey";
 pub const GET_STATUS_PATH: &str = "/status";

crates/common/src/pbs/error.rs

@@ -0,0 +1,69 @@
+use alloy::{
+    primitives::{B256, U256},
+    rpc::types::beacon::BlsPublicKey,
+};
+use thiserror::Error;
+
+use crate::error::BlstErrorWrapper;
+
+#[derive(Debug, Error)]
+pub enum PbsError {
+    #[error("axum error: {0}")]
+    AxumError(#[from] axum::Error),
+
+    #[error("reqwest error: {0}")]
+    Reqwest(#[from] reqwest::Error),
+
+    #[error("serde decode error: {0}")]
+    SerdeDecodeError(#[from] serde_json::Error),
+
+    #[error("relay response error. Code: {code}, err: {error_msg}")]
+    RelayResponse { error_msg: String, code: u16 },
+
+    #[error("failed validating relay response: {0}")]
+    Validation(#[from] ValidationError),
+
+    #[error("URL parsing error: {0}")]
+    UrlParsing(#[from] url::ParseError),
+}
+
+impl PbsError {
+    pub fn is_timeout(&self) -> bool {
+        matches!(self, PbsError::Reqwest(err) if err.is_timeout())
+    }
+}
+
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum ValidationError {
+    #[error("empty blockhash")]
+    EmptyBlockhash,
+
+    #[error("pubkey mismatch: expected {expected} got {got}")]
+    PubkeyMismatch { expected: BlsPublicKey, got: BlsPublicKey },
+
+    #[error("parent hash mismatch: expected {expected} got {got}")]
+    ParentHashMismatch { expected: B256, got: B256 },
+
+    #[error("block hash mismatch: expected {expected} got {got}")]
+    BlockHashMismatch { expected: B256, got: B256 },
+
+    #[error("mismatch in KZG commitments: exepcted_blobs: {expected_blobs} got_blobs: {got_blobs} got_commitments: {got_commitments} got_proofs: {got_proofs}")]
+    KzgCommitments {
+        expected_blobs: usize,
+        got_blobs: usize,
+        got_commitments: usize,
+        got_proofs: usize,
+    },
+
+    #[error("mismatch in KZG blob commitment: expected: {expected} got: {got} index: {index}")]
+    KzgMismatch { expected: String, got: String, index: usize },
+
+    #[error("bid below minimum: min: {min} got {got}")]
+    BidTooLow { min: U256, got: U256 },
+
+    #[error("empty tx root")]
+    EmptyTxRoot,
+
+    #[error("failed signature verification: {0:?}")]
+    Sigverify(#[from] BlstErrorWrapper),
+}

crates/common/src/pbs/mod.rs

@@ -1,4 +1,5 @@
 mod constants;
+pub mod error;
 mod event;
 mod relay;
 mod types;

crates/common/src/pbs/relay.rs

@@ -4,33 +4,34 @@ use alloy::{
     primitives::{hex::FromHex, B256},
     rpc::types::beacon::BlsPublicKey,
 };
-use eyre::{Result, WrapErr};
+use eyre::WrapErr;
 use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
 use serde::{Deserialize, Serialize};
 use url::Url;
 
 use super::{
-    constants::{BULDER_API_PATH, GET_STATUS_PATH, REGISTER_VALIDATOR_PATH, SUBMIT_BLOCK_PATH},
+    constants::{BUILDER_API_PATH, GET_STATUS_PATH, REGISTER_VALIDATOR_PATH, SUBMIT_BLOCK_PATH},
+    error::PbsError,
     HEADER_VERSION_KEY, HEADER_VERSION_VALUE,
 };
 use crate::{config::RelayConfig, DEFAULT_REQUEST_TIMEOUT};
 /// A parsed entry of the relay url in the format: scheme://pubkey@host
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Clone)]
 pub struct RelayEntry {
     /// Default if of the relay, the hostname of the url
     pub id: String,
     /// Public key of the relay
     pub pubkey: BlsPublicKey,
     /// Full url of the relay
-    pub url: String,
+    pub url: Url,
 }
 
 impl Serialize for RelayEntry {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
     {
-        serializer.serialize_str(&self.url)
+        self.url.serialize(serializer)
     }
 }
 
@@ -39,12 +40,11 @@ impl<'de> Deserialize<'de> for RelayEntry {
     where
         D: serde::Deserializer<'de>,
     {
-        let str = String::deserialize(deserializer)?;
-        let url = Url::parse(&str).map_err(serde::de::Error::custom)?;
+        let url = Url::deserialize(deserializer)?;
         let pubkey = BlsPublicKey::from_hex(url.username()).map_err(serde::de::Error::custom)?;
         let id = url.host().ok_or(serde::de::Error::custom("missing host"))?.to_string();
 
-        Ok(RelayEntry { pubkey, url: str, id })
+        Ok(RelayEntry { pubkey, url, id })
     }
 }
 
@@ -60,7 +60,7 @@ pub struct RelayClient {
 }
 
 impl RelayClient {
-    pub fn new(config: RelayConfig) -> Result<Self> {
+    pub fn new(config: RelayConfig) -> eyre::Result<Self> {
         let mut headers = HeaderMap::new();
         headers.insert(HEADER_VERSION_KEY, HeaderValue::from_static(HEADER_VERSION_VALUE));
 
@@ -90,46 +90,87 @@ impl RelayClient {
     }
 
     // URL builders
-    pub fn get_url(&self, path: &str) -> String {
-        format!("{}{path}", &self.config.entry.url)
+    pub fn get_url(&self, path: &str) -> Result<Url, PbsError> {
+        self.config.entry.url.join(path).map_err(PbsError::UrlParsing)
+    }
+    pub fn builder_api_url(&self, path: &str) -> Result<Url, PbsError> {
+        self.get_url(&format!("{BUILDER_API_PATH}{path}"))
     }
 
     pub fn get_header_url(
         &self,
         slot: u64,
         parent_hash: B256,
         validator_pubkey: BlsPublicKey,
-    ) -> String {
-        self.get_url(&format!("{BULDER_API_PATH}/header/{slot}/{parent_hash}/{validator_pubkey}"))
+    ) -> Result<Url, PbsError> {
+        self.builder_api_url(&format!("/header/{slot}/{parent_hash}/{validator_pubkey}"))
     }
 
-    pub fn get_status_url(&self) -> String {
-        self.get_url(&format!("{BULDER_API_PATH}{GET_STATUS_PATH}"))
+    pub fn get_status_url(&self) -> Result<Url, PbsError> {
+        self.builder_api_url(GET_STATUS_PATH)
     }
 
-    pub fn register_validator_url(&self) -> String {
-        self.get_url(&format!("{BULDER_API_PATH}{REGISTER_VALIDATOR_PATH}"))
+    pub fn register_validator_url(&self) -> Result<Url, PbsError> {
+        self.builder_api_url(REGISTER_VALIDATOR_PATH)
     }
 
-    pub fn submit_block_url(&self) -> String {
-        self.get_url(&format!("{BULDER_API_PATH}{SUBMIT_BLOCK_PATH}"))
+    pub fn submit_block_url(&self) -> Result<Url, PbsError> {
+        self.builder_api_url(SUBMIT_BLOCK_PATH)
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use alloy::{primitives::hex::FromHex, rpc::types::beacon::BlsPublicKey};
+    use alloy::{
+        primitives::{hex::FromHex, B256},
+        rpc::types::beacon::BlsPublicKey,
+    };
 
-    use super::RelayEntry;
+    use super::{RelayClient, RelayEntry};
+    use crate::config::RelayConfig;
 
     #[test]
     fn test_relay_entry() {
-        let s = "http://0xac6e77dfe25ecd6110b8e780608cce0dab71fdd5ebea22a16c0205200f2f8e2e3ad3b71d3499c54ad14d6c21b41a37ae@abc.xyz";
+        let s = "http://0xac6e77dfe25ecd6110b8e780608cce0dab71fdd5ebea22a16c0205200f2f8e2e3ad3b71d3499c54ad14d6c21b41a37ae@abc.xyz/";
 
         let parsed = serde_json::from_str::<RelayEntry>(&format!("\"{s}\"")).unwrap();
 
         assert_eq!(parsed.pubkey, BlsPublicKey::from_hex("0xac6e77dfe25ecd6110b8e780608cce0dab71fdd5ebea22a16c0205200f2f8e2e3ad3b71d3499c54ad14d6c21b41a37ae").unwrap());
-        assert_eq!(parsed.url, s);
+        assert_eq!(parsed.url.as_str(), s);
         assert_eq!(parsed.id, "abc.xyz");
     }
+
+    #[test]
+    fn test_relay_url() {
+        let slot = 0;
+        let parent_hash = B256::ZERO;
+        let validator_pubkey = BlsPublicKey::ZERO;
+        let expected = format!("http://0xa1cec75a3f0661e99299274182938151e8433c61a19222347ea1313d839229cb4ce4e3e5aa2bdeb71c8fcf1b084963c2@abc.xyz/eth/v1/builder/header/{slot}/{parent_hash}/{validator_pubkey}");
+
+        let relay_config = r#"
+        {
+            "url": "http://0xa1cec75a3f0661e99299274182938151e8433c61a19222347ea1313d839229cb4ce4e3e5aa2bdeb71c8fcf1b084963c2@abc.xyz"
+        }"#;
+
+        let config = serde_json::from_str::<RelayConfig>(relay_config).unwrap();
+        let relay = RelayClient::new(config).unwrap();
+
+        assert_eq!(
+            relay.get_header_url(slot, parent_hash, validator_pubkey).unwrap().to_string(),
+            expected
+        );
+
+        let relay_config = r#"
+        {
+            "url": "http://0xa1cec75a3f0661e99299274182938151e8433c61a19222347ea1313d839229cb4ce4e3e5aa2bdeb71c8fcf1b084963c2@abc.xyz//"
+        }"#;
+
+        let config = serde_json::from_str::<RelayConfig>(relay_config).unwrap();
+        let relay = RelayClient::new(config).unwrap();
+
+        assert_eq!(
+            relay.get_header_url(slot, parent_hash, validator_pubkey).unwrap().to_string(),
+            expected
+        );
+    }
 }

crates/pbs/Cargo.toml

@@ -34,5 +34,6 @@ blst.workspace = true
 # misc
 thiserror.workspace = true
 eyre.workspace = true
+url.workspace = true
 uuid.workspace = true
 lazy_static.workspace = true