Merge branch 'master' into master-sf

beacon_node/Cargo.toml

@@ -41,3 +41,4 @@ eth2-libp2p = { path = "./eth2-libp2p" }
 eth2_ssz = { path = "../eth2/utils/ssz" }
 toml = "0.5.4"
 serde = "1.0.102"
+clap_utils = { path = "../eth2/utils/clap_utils" }

beacon_node/client/src/notifier.rs

@@ -147,6 +147,7 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
                 );
             } else {
                 if sync_state.is_synced() {
+                    let block_info = if current_slot > head_slot { format!("   …  empty") } else { format!("{}", head_root) };
                     info!(
                         log_2,
                         "Synced";
@@ -154,6 +155,7 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
                         "finalized_root" => format!("{}", finalized_root),
                         "finalized_epoch" => finalized_epoch,
                         "epoch" => current_epoch,
+                        "block" => block_info,
                         "slot" => current_slot,
                     );
                 } else {

beacon_node/eth2-libp2p/src/config.rs

@@ -92,7 +92,7 @@ impl Default for Config {
         let gossip_message_id = |message: &GossipsubMessage| {
             MessageId(base64::encode_config(
                 &Sha256::digest(&message.data),
-                base64::URL_SAFE,
+                base64::URL_SAFE_NO_PAD,
             ))
         };
 

beacon_node/genesis/src/eth1_genesis_service.rs

@@ -5,7 +5,7 @@ use eth1::{DepositLog, Eth1Block, Service};
 use parking_lot::Mutex;
 use slog::{debug, error, info, trace, Logger};
 use state_processing::{
-    initialize_beacon_state_from_eth1, is_valid_genesis_state,
+    eth2_genesis_time, initialize_beacon_state_from_eth1, is_valid_genesis_state,
     per_block_processing::process_deposit, process_activations,
 };
 use std::sync::Arc;
@@ -186,9 +186,15 @@ impl Eth1GenesisService {
             .blocks()
             .read()
             .iter()
-            // It's only worth scanning blocks that have timestamps _after_ genesis time. It's
-            // impossible for any other block to trigger genesis.
-            .filter(|block| block.timestamp >= spec.min_genesis_time)
+            // Filter out any blocks that would result in a genesis time that is earlier than
+            // `MIN_GENESIS_TIME`.
+            //
+            // Note: any `SafeArith` errors are suppressed here; we simply skip blocks that cause
+            // overflow/div-by-zero.
+            .filter(|block| {
+                eth2_genesis_time(block.timestamp, spec)
+                    .map_or(false, |t| t >= spec.min_genesis_time)
+            })
             // The block cache might be more recently updated than deposit cache. Restrict any
             // block numbers that are not known by all caches.
             .filter(|block| {

beacon_node/network/src/attestation_service/mod.rs

@@ -10,7 +10,7 @@ use rand::seq::SliceRandom;
 use rest_types::ValidatorSubscription;
 use slog::{crit, debug, error, o, warn};
 use slot_clock::SlotClock;
-use std::collections::{HashMap, VecDeque};
+use std::collections::VecDeque;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use types::{Attestation, EthSpec, Slot, SubnetId};
@@ -77,8 +77,8 @@ pub struct AttestationService<T: BeaconChainTypes> {
     /// A collection of timeouts for when to unsubscribe from a shard subnet.
     unsubscriptions: HashSetDelay<ExactSubnet>,
 
-    /// A mapping indicating the number of known aggregate validators for a given `ExactSubnet`.
-    _aggregate_validators_on_subnet: HashMap<ExactSubnet, usize>,
+    /// A collection timeouts to track the existence of aggregate validator subscriptions at an `ExactSubnet`.
+    aggregate_validators_on_subnet: HashSetDelay<ExactSubnet>,
 
     /// A collection of seen validators. These dictate how many random subnets we should be
     /// subscribed to. As these time out, we unsubscribe for the required random subnets and update
@@ -124,7 +124,7 @@ impl<T: BeaconChainTypes> AttestationService<T> {
             discover_peers: HashSetDelay::new(default_timeout),
             subscriptions: HashSetDelay::new(default_timeout),
             unsubscriptions: HashSetDelay::new(default_timeout),
-            _aggregate_validators_on_subnet: HashMap::new(),
+            aggregate_validators_on_subnet: HashSetDelay::new(default_timeout),
             known_validators: HashSetDelay::new(last_seen_val_timeout),
             log,
         }
@@ -176,10 +176,12 @@ impl<T: BeaconChainTypes> AttestationService<T> {
             // sophisticated logic should be added using known future forks.
             // TODO: Implement
 
-            // set the subscription timer to subscribe to the next subnet if required
-            if let Err(e) = self.subscribe_to_subnet(exact_subnet) {
-                warn!(self.log, "Subscription to subnet error"; "error" => e);
-                return Err(());
+            if subscription.is_aggregator {
+                // set the subscription timer to subscribe to the next subnet if required
+                if let Err(e) = self.subscribe_to_subnet(exact_subnet) {
+                    warn!(self.log, "Subscription to subnet error"; "error" => e);
+                    return Err(());
+                }
             }
         }
         Ok(())
@@ -208,8 +210,11 @@ impl<T: BeaconChainTypes> AttestationService<T> {
             return false;
         }
 
-        // TODO: Correctly handle validation aggregator checks
-        true
+        let exact_subnet = ExactSubnet {
+            subnet_id: subnet.clone(),
+            slot: attestation.data.slot,
+        };
+        self.aggregate_validators_on_subnet.contains(&exact_subnet)
     }
 
     /* Internal private functions */
@@ -335,6 +340,12 @@ impl<T: BeaconChainTypes> AttestationService<T> {
             }
         };
 
+        // Regardless of whether or not we have already subscribed to a subnet, track the expiration
+        // of aggregate validator subscriptions to exact subnets so we know whether or not to drop
+        // attestations for a given subnet + slot
+        self.aggregate_validators_on_subnet
+            .insert_at(exact_subnet.clone(), expected_end_subscription_duration);
+
         // Checks on current subscriptions
         // Note: We may be connected to a long-lived random subnet. In this case we still add the
         // subscription timeout and check this case when the timeout fires. This is because a
@@ -641,6 +652,8 @@ impl<T: BeaconChainTypes> Stream for AttestationService<T> {
                 {
                     let _ = self.handle_known_validator_expiry();
                 }
+        // poll to remove entries on expiration, no need to act on expiration events
+        let _ = self.aggregate_validators_on_subnet.poll().map_err(|e| { error!(self.log, "Failed to check for aggregate validator on subnet expirations"; "error"=> format!("{}", e)); });
 
         // process any generated events
         if let Some(event) = self.events.pop_front() {

beacon_node/src/config.rs

@@ -1,5 +1,6 @@
 use beacon_chain::builder::PUBKEY_CACHE_FILENAME;
 use clap::ArgMatches;
+use clap_utils::BAD_TESTNET_DIR_MESSAGE;
 use client::{config::DEFAULT_DATADIR, ClientConfig, ClientGenesis};
 use eth2_libp2p::{Enr, Multiaddr};
 use eth2_testnet_config::Eth2TestnetConfig;
@@ -385,14 +386,8 @@ pub fn get_eth2_testnet_config<E: EthSpec>(
         Eth2TestnetConfig::load(testnet_dir.clone())
             .map_err(|e| format!("Unable to open testnet dir at {:?}: {}", testnet_dir, e))?
     } else {
-        Eth2TestnetConfig::hard_coded().map_err(|e| {
-            format!(
-                "The hard-coded testnet directory was invalid. \
-                 This happens when Lighthouse is migrating between spec versions. \
-                 Error : {}",
-                e
-            )
-        })?
+        Eth2TestnetConfig::hard_coded()
+            .map_err(|e| format!("{} Error : {}", BAD_TESTNET_DIR_MESSAGE, e))?
     })
 }