feat(mux): add mux file loader (#192)

* add mux file loader

* fix

Author: ltitanb

config.example.toml

@@ -103,11 +103,17 @@ frequency_get_header_ms = 300
 # - the mux is only used for get header requests
 # - if any value is missing from the mux config, the default value from the main config will be used
 [[mux]]
-# Which validator pubkeys to match against this mux config
+# Unique ID for the mux config
+id = "test_mux"
+# Which validator pubkeys to match against this mux config. This can be empty or omitted if a loader is specified. 
+# Any keys loaded via the loader will be added to this list.
 validator_pubkeys = [
     "0x80c7f782b2467c5898c5516a8b6595d75623960b4afc4f71ee07d40985d20e117ba35e7cd352a3e75fb85a8668a3b745",
     "0xa119589bb33ef52acbb8116832bec2b58fca590fe5c85eac5d3230b44d5bc09fe73ccd21f88eab31d6de16194d17782e",
 ]
+# Path to a file containing a list of validator pubkeys
+# OPTIONAL
+loader = "./mux_keys.example.json"
 timeout_get_header_ms = 900
 late_in_slot_time_ms = 1500
 # For each mux, one or more [[pbs_mux.relays]] can be defined, which will be used for the matching validator pubkeys

configs/pbs-mux.toml

@@ -18,9 +18,11 @@ enable_timing_games = true
 target_first_request_ms = 200
 
 [[mux]]
+id = "test_mux"
 validator_pubkeys = [
     "0x80c7f782b2467c5898c5516a8b6595d75623960b4afc4f71ee07d40985d20e117ba35e7cd352a3e75fb85a8668a3b745",
 ]
+loader = "./mux_keys.example.json"
 timeout_get_header_ms = 900
 late_in_slot_time_ms = 1500
 

crates/cli/src/docker_init.rs

@@ -219,6 +219,17 @@ pub fn handle_docker_init(config_path: String, output_dir: String) -> Result<()>
     }
 
     let mut pbs_envs = IndexMap::from([get_env_val(CONFIG_ENV, CONFIG_DEFAULT)]);
+    let mut pbs_volumes = vec![config_volume.clone()];
+
+    if let Some(mux_config) = cb_config.muxes {
+        for mux in mux_config.muxes.iter() {
+            if let Some((env_name, actual_path, internal_path)) = mux.loader_env() {
+                let (key, val) = get_env_val(&env_name, &internal_path);
+                pbs_envs.insert(key, val);
+                pbs_volumes.push(Volumes::Simple(format!("{}:{}:ro", actual_path, internal_path)));
+            }
+        }
+    }
 
     if let Some((key, val)) = chain_spec_env.clone() {
         pbs_envs.insert(key, val);
@@ -251,7 +262,6 @@ pub fn handle_docker_init(config_path: String, output_dir: String) -> Result<()>
     pbs_envs.insert(key, val);
 
     // volumes
-    let mut pbs_volumes = vec![config_volume.clone()];
     pbs_volumes.extend(chain_spec_volume.clone());
     pbs_volumes.extend(get_log_volume(&cb_config.logs, PBS_MODULE_NAME));
 

crates/common/Cargo.toml

@@ -7,7 +7,7 @@ publish = false
 
 [dependencies]
 # ethereum
-alloy = { workspace = true }
+alloy.workspace = true
 ssz_types.workspace = true
 ethereum_serde_utils.workspace = true
 

crates/common/src/config/constants.rs

@@ -25,6 +25,8 @@ pub const BUILDER_URLS_ENV: &str = "CB_BUILDER_URLS";
 /// Where to receive BuilderAPI calls from beacon node
 pub const PBS_ENDPOINT_ENV: &str = "CB_PBS_ENDPOINT";
 
+pub const MUX_PATH_ENV: &str = "CB_MUX_PATH";
+
 ///////////////////////// SIGNER /////////////////////////
 
 pub const SIGNER_IMAGE_DEFAULT: &str = "ghcr.io/commit-boost/signer:latest";

crates/common/src/config/mux.rs

@@ -1,16 +1,17 @@
 use std::{
     collections::{HashMap, HashSet},
+    path::{Path, PathBuf},
     sync::Arc,
 };
 
 use alloy::rpc::types::beacon::BlsPublicKey;
-use eyre::{bail, ensure, eyre};
+use eyre::{bail, ensure, eyre, Context};
 use serde::{Deserialize, Serialize};
 
-use super::{PbsConfig, RelayConfig};
+use super::{load_optional_env_var, PbsConfig, RelayConfig, MUX_PATH_ENV};
 use crate::pbs::{RelayClient, RelayEntry};
 
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize)]
 pub struct PbsMuxes {
     /// List of PBS multiplexers
     #[serde(rename = "mux")]
@@ -19,6 +20,7 @@ pub struct PbsMuxes {
 
 #[derive(Debug, Clone)]
 pub struct RuntimeMuxConfig {
+    pub id: String,
     pub config: Arc<PbsConfig>,
     pub relays: Vec<RelayClient>,
 }
@@ -29,9 +31,18 @@ impl PbsMuxes {
         default_pbs: &PbsConfig,
         default_relays: &[RelayConfig],
     ) -> eyre::Result<HashMap<BlsPublicKey, RuntimeMuxConfig>> {
+        let mut muxes = self.muxes;
+
+        for mux in muxes.iter_mut() {
+            if let Some(loader) = &mux.loader {
+                let extra_keys = loader.load(&mux.id)?;
+                mux.validator_pubkeys.extend(extra_keys);
+            }
+        }
+
         // check that validator pubkeys are in disjoint sets
         let mut unique_pubkeys = HashSet::new();
-        for mux in self.muxes.iter() {
+        for mux in muxes.iter() {
             for pubkey in mux.validator_pubkeys.iter() {
                 if !unique_pubkeys.insert(pubkey) {
                     bail!("duplicate validator pubkey in muxes: {pubkey}");
@@ -41,11 +52,12 @@ impl PbsMuxes {
 
         let mut configs = HashMap::new();
         // fill the configs using the default pbs config and relay entries
-        for mux in self.muxes {
-            ensure!(!mux.relays.is_empty(), "mux config must have at least one relay");
+        for mux in muxes {
+            ensure!(!mux.relays.is_empty(), "mux config {} must have at least one relay", mux.id);
             ensure!(
                 !mux.validator_pubkeys.is_empty(),
-                "mux config must have at least one validator pubkey"
+                "mux config {} must have at least one validator pubkey",
+                mux.id
             );
 
             let mut relay_clients = Vec::with_capacity(mux.relays.len());
@@ -89,7 +101,7 @@ impl PbsMuxes {
             };
             let config = Arc::new(config);
 
-            let runtime_config = RuntimeMuxConfig { config, relays: relay_clients };
+            let runtime_config = RuntimeMuxConfig { id: mux.id, config, relays: relay_clients };
             for pubkey in mux.validator_pubkeys.iter() {
                 configs.insert(*pubkey, runtime_config.clone());
             }
@@ -100,16 +112,36 @@ impl PbsMuxes {
 }
 
 /// Configuration for the PBS Multiplexer
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize)]
 pub struct MuxConfig {
+    /// Identifier for this mux config
+    pub id: String,
     /// Relays to use for this mux config
     pub relays: Vec<PartialRelayConfig>,
     /// Which validator pubkeys to match against this mux config
+    #[serde(default)]
     pub validator_pubkeys: Vec<BlsPublicKey>,
+    /// Loader for extra validator pubkeys
+    pub loader: Option<MuxKeysLoader>,
     pub timeout_get_header_ms: Option<u64>,
     pub late_in_slot_time_ms: Option<u64>,
 }
 
+impl MuxConfig {
+    /// Returns the env, actual path, and internal path to use for the loader
+    pub fn loader_env(&self) -> Option<(String, String, String)> {
+        self.loader.as_ref().map(|loader| match loader {
+            MuxKeysLoader::File(path_buf) => {
+                let path =
+                    path_buf.to_str().unwrap_or_else(|| panic!("invalid path: {:?}", path_buf));
+                let internal_path = get_mux_path(&self.id);
+
+                (get_mux_env(&self.id), path.to_owned(), internal_path)
+            }
+        })
+    }
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 /// A relay config with all optional fields. See [`RelayConfig`] for the
 /// description of the fields.
@@ -136,3 +168,39 @@ impl PartialRelayConfig {
         }
     }
 }
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+#[serde(untagged)]
+pub enum MuxKeysLoader {
+    /// A file containing a list of validator pubkeys
+    File(PathBuf),
+}
+
+impl MuxKeysLoader {
+    pub fn load(&self, mux_id: &str) -> eyre::Result<Vec<BlsPublicKey>> {
+        match self {
+            Self::File(config_path) => {
+                // First try loading from env
+                let path: PathBuf = load_optional_env_var(&get_mux_env(mux_id))
+                    .map(PathBuf::from)
+                    .unwrap_or(config_path.clone());
+                let file = load_file(path)?;
+                serde_json::from_str(&file).wrap_err("failed to parse mux keys file")
+            }
+        }
+    }
+}
+
+fn load_file<P: AsRef<Path> + std::fmt::Debug>(path: P) -> eyre::Result<String> {
+    std::fs::read_to_string(&path).wrap_err(format!("Unable to find mux keys file: {path:?}"))
+}
+
+/// A different env var for each mux
+fn get_mux_env(mux_id: &str) -> String {
+    format!("{MUX_PATH_ENV}_{mux_id}")
+}
+
+/// Path to the mux file
+fn get_mux_path(mux_id: &str) -> String {
+    format!("/{mux_id}-mux_keys.json")
+}

crates/pbs/src/mev_boost/get_header.rs

@@ -51,12 +51,12 @@ pub async fn get_header<S: BuilderApiState>(
     }
 
     let ms_into_slot = ms_into_slot(params.slot, state.config.chain);
-    let (pbs_config, relays, is_mux) = state.mux_config_and_relays(&params.pubkey);
+    let (pbs_config, relays, maybe_mux_id) = state.mux_config_and_relays(&params.pubkey);
 
-    if is_mux {
-        debug!(pubkey = %params.pubkey, relays = relays.len(), "using mux config");
+    if let Some(mux_id) = maybe_mux_id {
+        debug!(mux_id, relays = relays.len(), pubkey = %params.pubkey, "using mux config");
     } else {
-        debug!(pubkey = %params.pubkey, relays = relays.len(), "using default config");
+        debug!(relays = relays.len(), pubkey = %params.pubkey, "using default config");
     }
 
     let max_timeout_ms = pbs_config

crates/pbs/src/state.rs

@@ -88,10 +88,10 @@ where
     pub fn mux_config_and_relays(
         &self,
         pubkey: &BlsPublicKey,
-    ) -> (&PbsConfig, &[RelayClient], bool) {
+    ) -> (&PbsConfig, &[RelayClient], Option<&str>) {
         match self.config.muxes.as_ref().and_then(|muxes| muxes.get(pubkey)) {
-            Some(mux) => (&mux.config, mux.relays.as_slice(), true),
-            None => (self.pbs_config(), self.relays(), false),
+            Some(mux) => (&mux.config, mux.relays.as_slice(), Some(&mux.id)),
+            None => (self.pbs_config(), self.relays(), None),
         }
     }
 

docs/docs/get_started/running/binary.md

@@ -22,7 +22,8 @@ Modules need some environment variables to work correctly.
 
 ### PBS Module
 - `CB_BUILDER_URLS`: optional, comma-separated list of urls to `events` modules where to post builder events
-
+- `CB_PBS_ENDPOINT`: optional, override the endpoint where the PBS module will open the port for the beacon node
+- `CB_MUX_PATH_{ID}`: optional, override where to load mux validator keys for mux with id=\{ID\}
 ### Signer Module
 - `CB_JWTS`: required, comma-separated list of `MODULE_ID=JWT` to process signature requests
 - `CB_SIGNER_PORT`: required, port to open the signer server on

mux_keys.example.json

@@ -0,0 +1,5 @@
+[
+    "0x8160998addda06f2956e5d1945461f33dbc140486e972b96f341ebf2bdb553a0e3feb127451f5332dd9e33469d37ca67",
+    "0x87b5dc7f78b68a7b5e7f2e8b9c2115f968332cbf6fc2caaaaa2c9dc219a58206b72c924805f2278c58b55790a2c3bf17",
+    "0x89e2f50fe5cd07ed2ff0a01340b2f717aa65cced6d89a79fdecc1e924be5f4bbe75c11598bb9a53d307bb39b8223bc52"
+]

tests/tests/pbs_integration.rs

@@ -218,7 +218,11 @@ async fn test_mux() -> Result<()> {
     tokio::spawn(start_mock_relay_service(mock_state.clone(), port + 2));
 
     let mut config = to_pbs_config(chain, get_pbs_static_config(port), relays);
-    let mux = RuntimeMuxConfig { config: config.pbs_config.clone(), relays: vec![mux_relay] };
+    let mux = RuntimeMuxConfig {
+        id: String::from("test"),
+        config: config.pbs_config.clone(),
+        relays: vec![mux_relay],
+    };
 
     let validator_pubkey = blst_pubkey_to_alloy(&random_secret().sk_to_pk());