Merge remote-tracking branch 'origin/unstable' into jimmy/lh-2271-act…

…ivate-peerdas-at-fulu-fork-and-remove-eip7594_fork_epoch

Cargo.toml

@@ -33,6 +33,7 @@ members = [
     "common/eth2_network_config",
     "common/eth2_wallet_manager",
     "common/filesystem",
+    "common/health_metrics",
     "common/lighthouse_version",
     "common/lockfile",
     "common/logging",
@@ -252,6 +253,7 @@ filesystem = { path = "common/filesystem" }
 fork_choice = { path = "consensus/fork_choice" }
 genesis = { path = "beacon_node/genesis" }
 gossipsub = { path = "beacon_node/lighthouse_network/gossipsub/" }
+health_metrics = { path = "common/health_metrics" }
 http_api = { path = "beacon_node/http_api" }
 initialized_validators = { path = "validator_client/initialized_validators" }
 int_to_bytes = { path = "consensus/int_to_bytes" }

account_manager/src/validator/create.rs

@@ -6,14 +6,13 @@ use account_utils::{
 };
 use clap::{Arg, ArgAction, ArgMatches, Command};
 use clap_utils::FLAG_HEADER;
-use directory::{
-    ensure_dir_exists, parse_path_or_default_with_flag, DEFAULT_SECRET_DIR, DEFAULT_WALLET_DIR,
-};
+use directory::{parse_path_or_default_with_flag, DEFAULT_SECRET_DIR, DEFAULT_WALLET_DIR};
 use environment::Environment;
 use eth2_wallet_manager::WalletManager;
 use slashing_protection::{SlashingDatabase, SLASHING_PROTECTION_FILENAME};
 use std::ffi::OsStr;
 use std::fs;
+use std::fs::create_dir_all;
 use std::path::{Path, PathBuf};
 use types::EthSpec;
 use validator_dir::Builder as ValidatorDirBuilder;
@@ -156,8 +155,10 @@ pub fn cli_run<E: EthSpec>(
         ));
     }
 
-    ensure_dir_exists(&validator_dir)?;
-    ensure_dir_exists(&secrets_dir)?;
+    create_dir_all(&validator_dir)
+        .map_err(|e| format!("Could not create validator dir at {validator_dir:?}: {e:?}"))?;
+    create_dir_all(&secrets_dir)
+        .map_err(|e| format!("Could not create secrets dir at {secrets_dir:?}: {e:?}"))?;
 
     eprintln!("secrets-dir path {:?}", secrets_dir);
     eprintln!("wallets-dir path {:?}", wallet_base_dir);

account_manager/src/validator/recover.rs

@@ -5,10 +5,10 @@ use account_utils::eth2_keystore::{keypair_from_secret, Keystore, KeystoreBuilde
 use account_utils::{random_password, read_mnemonic_from_cli, STDIN_INPUTS_FLAG};
 use clap::{Arg, ArgAction, ArgMatches, Command};
 use clap_utils::FLAG_HEADER;
-use directory::ensure_dir_exists;
 use directory::{parse_path_or_default_with_flag, DEFAULT_SECRET_DIR};
 use eth2_wallet::bip39::Seed;
 use eth2_wallet::{recover_validator_secret_from_mnemonic, KeyType, ValidatorKeystores};
+use std::fs::create_dir_all;
 use std::path::PathBuf;
 use validator_dir::Builder as ValidatorDirBuilder;
 pub const CMD: &str = "recover";
@@ -91,8 +91,10 @@ pub fn cli_run(matches: &ArgMatches, validator_dir: PathBuf) -> Result<(), Strin
 
     eprintln!("secrets-dir path: {:?}", secrets_dir);
 
-    ensure_dir_exists(&validator_dir)?;
-    ensure_dir_exists(&secrets_dir)?;
+    create_dir_all(&validator_dir)
+        .map_err(|e| format!("Could not create validator dir at {validator_dir:?}: {e:?}"))?;
+    create_dir_all(&secrets_dir)
+        .map_err(|e| format!("Could not create secrets dir at {secrets_dir:?}: {e:?}"))?;
 
     eprintln!();
     eprintln!("WARNING: KEY RECOVERY CAN LEAD TO DUPLICATING VALIDATORS KEYS, WHICH CAN LEAD TO SLASHING.");

account_manager/src/wallet/mod.rs

@@ -5,7 +5,8 @@ pub mod recover;
 use crate::WALLETS_DIR_FLAG;
 use clap::{Arg, ArgAction, ArgMatches, Command};
 use clap_utils::FLAG_HEADER;
-use directory::{ensure_dir_exists, parse_path_or_default_with_flag, DEFAULT_WALLET_DIR};
+use directory::{parse_path_or_default_with_flag, DEFAULT_WALLET_DIR};
+use std::fs::create_dir_all;
 use std::path::PathBuf;
 
 pub const CMD: &str = "wallet";
@@ -44,7 +45,7 @@ pub fn cli_run(matches: &ArgMatches) -> Result<(), String> {
     } else {
         parse_path_or_default_with_flag(matches, WALLETS_DIR_FLAG, DEFAULT_WALLET_DIR)?
     };
-    ensure_dir_exists(&wallet_base_dir)?;
+    create_dir_all(&wallet_base_dir).map_err(|_| "Could not create wallet base dir")?;
 
     eprintln!("wallet-dir path: {:?}", wallet_base_dir);
 

beacon_node/beacon_chain/src/attestation_verification.rs

@@ -62,7 +62,7 @@ use tree_hash::TreeHash;
 use types::{
     Attestation, AttestationRef, BeaconCommittee, BeaconStateError::NoCommitteeFound, ChainSpec,
     CommitteeIndex, Epoch, EthSpec, Hash256, IndexedAttestation, SelectionProof,
-    SignedAggregateAndProof, Slot, SubnetId,
+    SignedAggregateAndProof, SingleAttestation, Slot, SubnetId,
 };
 
 pub use batch::{batch_verify_aggregated_attestations, batch_verify_unaggregated_attestations};
@@ -317,12 +317,22 @@ pub struct VerifiedUnaggregatedAttestation<'a, T: BeaconChainTypes> {
     attestation: AttestationRef<'a, T::EthSpec>,
     indexed_attestation: IndexedAttestation<T::EthSpec>,
     subnet_id: SubnetId,
+    validator_index: usize,
 }
 
 impl<T: BeaconChainTypes> VerifiedUnaggregatedAttestation<'_, T> {
     pub fn into_indexed_attestation(self) -> IndexedAttestation<T::EthSpec> {
         self.indexed_attestation
     }
+
+    pub fn single_attestation(&self) -> Option<SingleAttestation> {
+        Some(SingleAttestation {
+            committee_index: self.attestation.committee_index()? as usize,
+            attester_index: self.validator_index,
+            data: self.attestation.data().clone(),
+            signature: self.attestation.signature().clone(),
+        })
+    }
 }
 
 /// Custom `Clone` implementation is to avoid the restrictive trait bounds applied by the usual derive
@@ -1035,6 +1045,7 @@ impl<'a, T: BeaconChainTypes> VerifiedUnaggregatedAttestation<'a, T> {
             attestation,
             indexed_attestation,
             subnet_id,
+            validator_index: validator_index as usize,
         })
     }
 

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2035,10 +2035,30 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             |v| {
                 // This method is called for API and gossip attestations, so this covers all unaggregated attestation events
                 if let Some(event_handler) = self.event_handler.as_ref() {
+                    if event_handler.has_single_attestation_subscribers() {
+                        let current_fork = self
+                            .spec
+                            .fork_name_at_slot::<T::EthSpec>(v.attestation().data().slot);
+                        if current_fork.electra_enabled() {
+                            // I don't see a situation where this could return None. The upstream unaggregated attestation checks
+                            // should have already verified that this is an attestation with a single committee bit set.
+                            if let Some(single_attestation) = v.single_attestation() {
+                                event_handler.register(EventKind::SingleAttestation(Box::new(
+                                    single_attestation,
+                                )));
+                            }
+                        }
+                    }
+
                     if event_handler.has_attestation_subscribers() {
-                        event_handler.register(EventKind::Attestation(Box::new(
-                            v.attestation().clone_as_attestation(),
-                        )));
+                        let current_fork = self
+                            .spec
+                            .fork_name_at_slot::<T::EthSpec>(v.attestation().data().slot);
+                        if !current_fork.electra_enabled() {
+                            event_handler.register(EventKind::Attestation(Box::new(
+                                v.attestation().clone_as_attestation(),
+                            )));
+                        }
                     }
                 }
                 metrics::inc_counter(&metrics::UNAGGREGATED_ATTESTATION_PROCESSING_SUCCESSES);

beacon_node/beacon_chain/src/events.rs

@@ -8,6 +8,7 @@ const DEFAULT_CHANNEL_CAPACITY: usize = 16;
 
 pub struct ServerSentEventHandler<E: EthSpec> {
     attestation_tx: Sender<EventKind<E>>,
+    single_attestation_tx: Sender<EventKind<E>>,
     block_tx: Sender<EventKind<E>>,
     blob_sidecar_tx: Sender<EventKind<E>>,
     finalized_tx: Sender<EventKind<E>>,
@@ -37,6 +38,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
 
     pub fn new_with_capacity(log: Logger, capacity: usize) -> Self {
         let (attestation_tx, _) = broadcast::channel(capacity);
+        let (single_attestation_tx, _) = broadcast::channel(capacity);
         let (block_tx, _) = broadcast::channel(capacity);
         let (blob_sidecar_tx, _) = broadcast::channel(capacity);
         let (finalized_tx, _) = broadcast::channel(capacity);
@@ -56,6 +58,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
 
         Self {
             attestation_tx,
+            single_attestation_tx,
             block_tx,
             blob_sidecar_tx,
             finalized_tx,
@@ -90,6 +93,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .attestation_tx
                 .send(kind)
                 .map(|count| log_count("attestation", count)),
+            EventKind::SingleAttestation(_) => self
+                .single_attestation_tx
+                .send(kind)
+                .map(|count| log_count("single_attestation", count)),
             EventKind::Block(_) => self
                 .block_tx
                 .send(kind)
@@ -164,6 +171,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.attestation_tx.subscribe()
     }
 
+    pub fn subscribe_single_attestation(&self) -> Receiver<EventKind<E>> {
+        self.single_attestation_tx.subscribe()
+    }
+
     pub fn subscribe_block(&self) -> Receiver<EventKind<E>> {
         self.block_tx.subscribe()
     }
@@ -232,6 +243,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.attestation_tx.receiver_count() > 0
     }
 
+    pub fn has_single_attestation_subscribers(&self) -> bool {
+        self.single_attestation_tx.receiver_count() > 0
+    }
+
     pub fn has_block_subscribers(&self) -> bool {
         self.block_tx.receiver_count() > 0
     }