Add broadcast test to simulator

testing/node_test_rig/Cargo.toml

@@ -11,7 +11,7 @@ types = { workspace = true }
 tempfile = { workspace = true }
 eth2 = { workspace = true }
 validator_client = { workspace = true }
-validator_dir = { workspace = true }
+validator_dir = { workspace = true, features = ["insecure_keys"] }
 sensitive_url = { workspace = true }
 execution_layer = { workspace = true }
 tokio = { workspace = true }

testing/node_test_rig/src/lib.rs

@@ -21,7 +21,7 @@ pub use eth2;
 pub use execution_layer::test_utils::{
     Config as MockServerConfig, MockExecutionConfig, MockServer,
 };
-pub use validator_client::Config as ValidatorConfig;
+pub use validator_client::{ApiTopic, Config as ValidatorConfig};
 
 /// The global timeout for HTTP requests to the beacon node.
 const HTTP_TIMEOUT: Duration = Duration::from_secs(8);

testing/simulator/src/eth1_sim.rs

@@ -10,7 +10,8 @@ use futures::prelude::*;
 use node_test_rig::environment::RuntimeContext;
 use node_test_rig::{
     environment::{EnvironmentBuilder, LoggerConfig},
-    testing_client_config, testing_validator_config, ClientConfig, ClientGenesis, ValidatorFiles,
+    testing_client_config, testing_validator_config, ApiTopic, ClientConfig, ClientGenesis,
+    ValidatorFiles,
 };
 use rayon::prelude::*;
 use sensitive_url::SensitiveUrl;
@@ -159,10 +160,25 @@ pub fn run_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
                         validator_config.fee_recipient = Some(SUGGESTED_FEE_RECIPIENT.into());
                     }
                     println!("Adding validator client {}", i);
-                    network_1
-                        .add_validator_client(validator_config, i, files, i % 2 == 0)
-                        .await
-                        .expect("should add validator");
+
+                    // Enable broadcast on every 4th node.
+                    if i % 4 == 0 {
+                        validator_config.broadcast_topics = ApiTopic::all();
+                        let beacon_nodes = vec![i, (i + 1) % node_count];
+                        network_1
+                            .add_validator_client_with_fallbacks(
+                                validator_config,
+                                i,
+                                beacon_nodes,
+                                files,
+                            )
+                            .await
+                    } else {
+                        network_1
+                            .add_validator_client(validator_config, i, files, i % 2 == 0)
+                            .await
+                    }
+                    .expect("should add validator");
                 },
                 "vc",
             );

testing/simulator/src/local_network.rs

@@ -270,6 +270,48 @@ impl<E: EthSpec> LocalNetwork<E> {
         Ok(())
     }
 
+    pub async fn add_validator_client_with_fallbacks(
+        &self,
+        mut validator_config: ValidatorConfig,
+        validator_index: usize,
+        beacon_nodes: Vec<usize>,
+        validator_files: ValidatorFiles,
+    ) -> Result<(), String> {
+        let context = self
+            .context
+            .service_context(format!("validator_{}", validator_index));
+        let self_1 = self.clone();
+        let mut beacon_node_urls = vec![];
+        for beacon_node in beacon_nodes {
+            let socket_addr = {
+                let read_lock = self.beacon_nodes.read();
+                let beacon_node = read_lock
+                    .get(beacon_node)
+                    .ok_or_else(|| format!("No beacon node for index {}", beacon_node))?;
+                beacon_node
+                    .client
+                    .http_api_listen_addr()
+                    .expect("Must have http started")
+            };
+            let beacon_node = SensitiveUrl::parse(
+                format!("http://{}:{}", socket_addr.ip(), socket_addr.port()).as_str(),
+            )
+            .unwrap();
+            beacon_node_urls.push(beacon_node);
+        }
+
+        validator_config.beacon_nodes = beacon_node_urls;
+
+        let validator_client = LocalValidatorClient::production_with_insecure_keypairs(
+            context,
+            validator_config,
+            validator_files,
+        )
+        .await?;
+        self_1.validator_clients.write().push(validator_client);
+        Ok(())
+    }
+
     /// For all beacon nodes in `Self`, return a HTTP client to access each nodes HTTP API.
     pub fn remote_nodes(&self) -> Result<Vec<BeaconNodeHttpClient>, String> {
         let beacon_nodes = self.beacon_nodes.read();

validator_client/src/beacon_node_fallback.rs

@@ -714,11 +714,35 @@ impl<T: SlotClock, E: EthSpec> BeaconNodeFallback<T, E> {
 }
 
 /// Serves as a cue for `BeaconNodeFallback` to tell which requests need to be broadcasted.
-#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, EnumString, EnumVariantNames)]
+#[derive(Clone, Copy, Debug, PartialEq, Deserialize, Serialize, EnumString, EnumVariantNames)]
 #[strum(serialize_all = "kebab-case")]
 pub enum ApiTopic {
     Attestations,
     Blocks,
     Subscriptions,
     SyncCommittee,
 }
+
+impl ApiTopic {
+    pub fn all() -> Vec<ApiTopic> {
+        use ApiTopic::*;
+        vec![Attestations, Blocks, Subscriptions, SyncCommittee]
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use std::str::FromStr;
+    use strum::VariantNames;
+
+    #[test]
+    fn api_topic_all() {
+        let all = ApiTopic::all();
+        assert_eq!(all.len(), ApiTopic::VARIANTS.len());
+        assert!(ApiTopic::VARIANTS
+            .iter()
+            .map(|topic| ApiTopic::from_str(topic).unwrap())
+            .eq(all.into_iter()));
+    }
+}