merge queue: embarking unstable (256d904), #5235 and #4867 together

beacon_node/lighthouse_network/src/service/mod.rs

@@ -1226,22 +1226,39 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
                                 .publish(Topic::from(topic.clone()), data)
                             {
                                 Ok(_) => {
-                                    warn!(self.log, "Gossip message published on retry"; "topic" => topic_str);
-                                    if let Some(v) = metrics::get_int_counter(
+                                    debug!(
+                                        self.log,
+                                        "Gossip message published on retry";
+                                        "topic" => topic_str
+                                    );
+                                    metrics::inc_counter_vec(
                                         &metrics::GOSSIP_LATE_PUBLISH_PER_TOPIC_KIND,
                                         &[topic_str],
-                                    ) {
-                                        v.inc()
-                                    };
+                                    );
+                                }
+                                Err(PublishError::Duplicate) => {
+                                    debug!(
+                                        self.log,
+                                        "Gossip message publish ignored on retry";
+                                        "reason" => "duplicate",
+                                        "topic" => topic_str
+                                    );
+                                    metrics::inc_counter_vec(
+                                        &metrics::GOSSIP_FAILED_LATE_PUBLISH_PER_TOPIC_KIND,
+                                        &[topic_str],
+                                    );
                                 }
                                 Err(e) => {
-                                    warn!(self.log, "Gossip message publish failed on retry"; "topic" => topic_str, "error" => %e);
-                                    if let Some(v) = metrics::get_int_counter(
+                                    warn!(
+                                        self.log,
+                                        "Gossip message publish failed on retry";
+                                        "topic" => topic_str,
+                                        "error" => %e
+                                    );
+                                    metrics::inc_counter_vec(
                                         &metrics::GOSSIP_FAILED_LATE_PUBLISH_PER_TOPIC_KIND,
                                         &[topic_str],
-                                    ) {
-                                        v.inc()
-                                    };
+                                    );
                                 }
                             }
                         }

book/src/help_vc.md

@@ -26,6 +26,9 @@ FLAGS:
             but is only safe if slashing protection is enabled on the remote signer and is implemented correctly. DO NOT
             ENABLE THIS FLAG UNLESS YOU ARE CERTAIN THAT SLASHING PROTECTION IS ENABLED ON THE REMOTE SIGNER. YOU WILL
             GET SLASHED IF YOU USE THIS FLAG WITHOUT ENABLING WEB3SIGNER'S SLASHING PROTECTION.
+        --distributed
+            Enables functionality required for running the validator in a distributed validator cluster.
+
         --enable-doppelganger-protection
             If this flag is set, Lighthouse will delay startup for three epochs and monitor for messages on the network
             by any of the validators managed by this client. This will result in three (possibly four) epochs worth of

validator_client/Cargo.toml

@@ -10,6 +10,7 @@ path = "src/lib.rs"
 
 [dev-dependencies]
 tokio = { workspace = true }
+itertools = { workspace = true }
 
 [dependencies]
 tree_hash = { workspace = true }
@@ -51,7 +52,6 @@ ring = { workspace = true }
 rand = { workspace = true, features = ["small_rng"] }
 lighthouse_metrics = { workspace = true }
 lazy_static = { workspace = true }
-itertools = { workspace = true }
 monitoring_api = { workspace = true }
 sensitive_url = { workspace = true }
 task_executor = { workspace = true }

validator_client/src/cli.rs

@@ -145,6 +145,12 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                        future.")
                 .takes_value(false)
         )
+        .arg(
+            Arg::with_name("distributed")
+                .long("distributed")
+                .help("Enables functionality required for running the validator in a distributed validator cluster.")
+                .takes_value(false)
+        )
         /* REST API related arguments */
         .arg(
             Arg::with_name("http")

validator_client/src/config.rs

@@ -84,6 +84,8 @@ pub struct Config {
     pub builder_boost_factor: Option<u64>,
     /// If true, Lighthouse will prefer builder proposals, if available.
     pub prefer_builder_proposals: bool,
+    /// Whether we are running with distributed network support.
+    pub distributed: bool,
     pub web3_signer_keep_alive_timeout: Option<Duration>,
     pub web3_signer_max_idle_connections: Option<usize>,
 }
@@ -130,6 +132,7 @@ impl Default for Config {
             produce_block_v3: false,
             builder_boost_factor: None,
             prefer_builder_proposals: false,
+            distributed: false,
             web3_signer_keep_alive_timeout: Some(Duration::from_secs(90)),
             web3_signer_max_idle_connections: None,
         }
@@ -233,6 +236,10 @@ impl Config {
             config.beacon_nodes_tls_certs = Some(tls_certs.split(',').map(PathBuf::from).collect());
         }
 
+        if cli_args.is_present("distributed") {
+            config.distributed = true;
+        }
+
         if cli_args.is_present("disable-run-on-all") {
             warn!(
                 log,

validator_client/src/duties_service.rs

@@ -6,7 +6,7 @@
 //! The `DutiesService` is also responsible for sending events to the `BlockService` which trigger
 //! block production.
 
-mod sync;
+pub mod sync;
 
 use crate::beacon_node_fallback::{ApiTopic, BeaconNodeFallback, OfflineOnFailure, RequireSynced};
 use crate::http_metrics::metrics::{get_int_gauge, set_int_gauge, ATTESTATION_DUTY};
@@ -42,6 +42,9 @@ const HISTORICAL_DUTIES_EPOCHS: u64 = 2;
 /// At start-up selection proofs will be computed with less lookahead out of necessity.
 const SELECTION_PROOF_SLOT_LOOKAHEAD: u64 = 8;
 
+/// The attestation selection proof lookahead for those running with the --distributed flag.
+const SELECTION_PROOF_SLOT_LOOKAHEAD_DVT: u64 = 1;
+
 /// Fraction of a slot at which selection proof signing should happen (2 means half way).
 const SELECTION_PROOF_SCHEDULE_DENOM: u32 = 2;
 
@@ -211,16 +214,21 @@ pub struct DutiesService<T, E: EthSpec> {
     /// proposals for any validators which are not registered locally.
     pub proposers: RwLock<ProposerMap>,
     /// Map from validator index to sync committee duties.
-    pub sync_duties: SyncDutiesMap,
+    pub sync_duties: SyncDutiesMap<E>,
     /// Provides the canonical list of locally-managed validators.
     pub validator_store: Arc<ValidatorStore<T, E>>,
     /// Tracks the current slot.
     pub slot_clock: T,
     /// Provides HTTP access to remote beacon nodes.
     pub beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
-    pub enable_high_validator_count_metrics: bool,
+    /// The runtime for spawning tasks.
     pub context: RuntimeContext<E>,
+    /// The current chain spec.
     pub spec: ChainSpec,
+    //// Whether we permit large validator counts in the metrics.
+    pub enable_high_validator_count_metrics: bool,
+    /// If this validator is running in distributed mode.
+    pub distributed: bool,
 }
 
 impl<T: SlotClock + 'static, E: EthSpec> DutiesService<T, E> {
@@ -997,7 +1005,13 @@ async fn fill_in_selection_proofs<T: SlotClock + 'static, E: EthSpec>(
                 continue;
             };
 
-            let lookahead_slot = current_slot + SELECTION_PROOF_SLOT_LOOKAHEAD;
+            let selection_lookahead = if duties_service.distributed {
+                SELECTION_PROOF_SLOT_LOOKAHEAD_DVT
+            } else {
+                SELECTION_PROOF_SLOT_LOOKAHEAD
+            };
+
+            let lookahead_slot = current_slot + selection_lookahead;
 
             let mut relevant_duties = duties_by_slot.split_off(&lookahead_slot);
             std::mem::swap(&mut relevant_duties, &mut duties_by_slot);