Connects the attestation service to network components (#961)

* Sends attestations to the attestation service for processing

* Adds 'attnets' field to local ENR

* Adds ENR bitfield modification logic

* Link attestation service to discovery

- Updates discv5
- Links discover events to discovery
- Support for ENRBitfield

* Adds discovery config params, correct warnings

* Rust fmt fixes

* Correct tests

beacon_node/network/src/attestation_service/mod.rs

@@ -3,19 +3,17 @@
 //! determines whether attestations should be aggregated and/or passed to the beacon node.
 
 use beacon_chain::{BeaconChain, BeaconChainTypes};
-use eth2_libp2p::{types::GossipKind, NetworkGlobals};
+use eth2_libp2p::{types::GossipKind, MessageId, NetworkGlobals, PeerId};
 use futures::prelude::*;
 use hashmap_delay::HashSetDelay;
 use rand::seq::SliceRandom;
 use rest_types::ValidatorSubscription;
 use slog::{crit, debug, error, o, warn};
 use slot_clock::SlotClock;
-use std::boxed::Box;
 use std::collections::VecDeque;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
-use types::{Attestation, SubnetId};
-use types::{EthSpec, Slot};
+use types::{Attestation, EthSpec, SignedAggregateAndProof, Slot, SubnetId};
 
 /// The minimum number of slots ahead that we attempt to discover peers for a subscription. If the
 /// slot is less than this number, skip the peer discovery process.
@@ -44,6 +42,8 @@ pub enum AttServiceMessage {
     EnrRemove(SubnetId),
     /// Discover peers for a particular subnet.
     DiscoverPeers(SubnetId),
+    /// Propagate an attestation if it's deemed valid.
+    Propagate(PeerId, MessageId),
 }
 
 pub struct AttestationService<T: BeaconChainTypes> {
@@ -152,11 +152,29 @@ impl<T: BeaconChainTypes> AttestationService<T> {
         Ok(())
     }
 
-    pub fn handle_attestation(
+    /// Handles un-aggregated attestations from the network.
+    pub fn handle_unaggregated_attestation(
         &mut self,
+        message_id: MessageId,
+        peer_id: PeerId,
         subnet: SubnetId,
-        attestation: Box<Attestation<T::EthSpec>>,
+        attestation: Attestation<T::EthSpec>,
     ) {
+        // TODO: Handle attestation processing
+        self.events
+            .push_back(AttServiceMessage::Propagate(peer_id, message_id));
+    }
+
+    /// Handles aggregate attestations from the network.
+    pub fn handle_aggregate_attestation(
+        &mut self,
+        message_id: MessageId,
+        peer_id: PeerId,
+        attestation: SignedAggregateAndProof<T::EthSpec>,
+    ) {
+        // TODO: Handle attestation processing
+        self.events
+            .push_back(AttServiceMessage::Propagate(peer_id, message_id));
     }
 
     /* Internal private functions */
@@ -231,6 +249,12 @@ impl<T: BeaconChainTypes> AttestationService<T> {
                 self.discover_peers
                     .insert_at((subnet_id, subscription_slot), duration_to_discover);
             }
+        } else {
+            // TODO: Send the time frame needed to have a peer connected, so that we can
+            // maintain peers for a least this duration.
+            // We may want to check the global PeerInfo to see estimated timeouts for each
+            // peer before they can be removed.
+            return Err("Not enough time for a discovery search");
         }
         Ok(())
     }

beacon_node/network/src/attestation_service/process.rs

@@ -0,0 +1,55 @@
+
+    /// Process a gossip message declaring a new attestation.
+    ///
+    /// Not currently implemented.
+    pub fn on_attestation_gossip(&mut self, _peer_id: PeerId, _msg: Attestation<T::EthSpec>) {
+        // TODO: Handle subnet gossip
+        /*
+        match self.chain.process_attestation(msg.clone()) {
+            Ok(outcome) => match outcome {
+                AttestationProcessingOutcome::Processed => {
+                    debug!(
+                        self.log,
+                        "Processed attestation";
+                        "source" => "gossip",
+                        "peer" => format!("{:?}",peer_id),
+                        "block_root" => format!("{}", msg.data.beacon_block_root),
+                        "slot" => format!("{}", msg.data.slot),
+                    );
+                }
+                AttestationProcessingOutcome::UnknownHeadBlock { beacon_block_root } => {
+                    // TODO: Maintain this attestation and re-process once sync completes
+                    trace!(
+                    self.log,
+                    "Attestation for unknown block";
+                    "peer_id" => format!("{:?}", peer_id),
+                    "block" => format!("{}", beacon_block_root)
+                    );
+                    // we don't know the block, get the sync manager to handle the block lookup
+                    self.send_to_sync(SyncMessage::UnknownBlockHash(peer_id, beacon_block_root));
+                }
+                AttestationProcessingOutcome::FutureEpoch { .. }
+                | AttestationProcessingOutcome::PastEpoch { .. }
+                | AttestationProcessingOutcome::UnknownTargetRoot { .. }
+                | AttestationProcessingOutcome::FinalizedSlot { .. } => {} // ignore the attestation
+                AttestationProcessingOutcome::Invalid { .. }
+                | AttestationProcessingOutcome::EmptyAggregationBitfield { .. }
+                | AttestationProcessingOutcome::AttestsToFutureBlock { .. }
+                | AttestationProcessingOutcome::InvalidSignature
+                | AttestationProcessingOutcome::NoCommitteeForSlotAndIndex { .. }
+                | AttestationProcessingOutcome::BadTargetEpoch { .. } => {
+                    // the peer has sent a bad attestation. Remove them.
+                    self.network.disconnect(peer_id, GoodbyeReason::Fault);
+                }
+            },
+            Err(_) => {
+                // error is logged during the processing therefore no error is logged here
+                trace!(
+                    self.log,
+                    "Erroneous gossip attestation ssz";
+                    "ssz" => format!("0x{}", hex::encode(msg.as_ssz_bytes())),
+                );
+            }
+        };
+        */
+    }

beacon_node/network/src/router/mod.rs

@@ -16,7 +16,7 @@ use eth2_libp2p::{
 use futures::future::Future;
 use futures::stream::Stream;
 use processor::Processor;
-use slog::{debug, o, trace, warn};
+use slog::{crit, debug, o, trace, warn};
 use std::sync::Arc;
 use tokio::sync::mpsc;
 use types::EthSpec;
@@ -224,19 +224,6 @@ impl<T: BeaconChainTypes> Router<T> {
                 }
                 self.processor.on_block_gossip(peer_id, block);
             }
-            PubsubData::AggregateAndProofAttestation(_agg_attestation) => {
-                // TODO: Handle propagation conditions
-                self.propagate_message(id, peer_id);
-                // TODO Handle aggregate attestion
-                // self.processor
-                //    .on_attestation_gossip(peer_id.clone(), &agg_attestation);
-            }
-            PubsubData::Attestation(boxed_shard_attestation) => {
-                // TODO: Handle propagation conditions
-                self.propagate_message(id, peer_id.clone());
-                self.processor
-                    .on_attestation_gossip(peer_id, boxed_shard_attestation.1);
-            }
             PubsubData::VoluntaryExit(_exit) => {
                 // TODO: Apply more sophisticated validation
                 self.propagate_message(id, peer_id.clone());
@@ -255,6 +242,19 @@ impl<T: BeaconChainTypes> Router<T> {
                 // TODO: Handle attester slashings
                 debug!(self.log, "Received an attester slashing"; "peer_id" => format!("{}", peer_id) );
             }
+            // Attestations should never reach the router.
+            PubsubData::AggregateAndProofAttestation(_agg_attestation) => {
+                crit!(
+                    self.log,
+                    "Attestations should always be handled by the attestation service"
+                );
+            }
+            PubsubData::Attestation(_boxed_subnet_attestation) => {
+                crit!(
+                    self.log,
+                    "Attestations should always be handled by the attestation service"
+                );
+            }
         }
     }
 

beacon_node/network/src/router/processor.rs

@@ -1,8 +1,6 @@
 use crate::service::NetworkMessage;
 use crate::sync::SyncMessage;
-use beacon_chain::{
-    AttestationProcessingOutcome, BeaconChain, BeaconChainTypes, BlockProcessingOutcome,
-};
+use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessingOutcome};
 use eth2_libp2p::rpc::methods::*;
 use eth2_libp2p::rpc::{RPCEvent, RPCRequest, RPCResponse, RequestId};
 use eth2_libp2p::PeerId;
@@ -553,61 +551,6 @@ impl<T: BeaconChainTypes> Processor<T> {
         // TODO: Update with correct block gossip checking
         true
     }
-
-    /// Process a gossip message declaring a new attestation.
-    ///
-    /// Not currently implemented.
-    pub fn on_attestation_gossip(&mut self, _peer_id: PeerId, _msg: Attestation<T::EthSpec>) {
-        // TODO: Handle subnet gossip
-        /*
-        match self.chain.process_attestation(msg.clone()) {
-            Ok(outcome) => match outcome {
-                AttestationProcessingOutcome::Processed => {
-                    debug!(
-                        self.log,
-                        "Processed attestation";
-                        "source" => "gossip",
-                        "peer" => format!("{:?}",peer_id),
-                        "block_root" => format!("{}", msg.data.beacon_block_root),
-                        "slot" => format!("{}", msg.data.slot),
-                    );
-                }
-                AttestationProcessingOutcome::UnknownHeadBlock { beacon_block_root } => {
-                    // TODO: Maintain this attestation and re-process once sync completes
-                    trace!(
-                    self.log,
-                    "Attestation for unknown block";
-                    "peer_id" => format!("{:?}", peer_id),
-                    "block" => format!("{}", beacon_block_root)
-                    );
-                    // we don't know the block, get the sync manager to handle the block lookup
-                    self.send_to_sync(SyncMessage::UnknownBlockHash(peer_id, beacon_block_root));
-                }
-                AttestationProcessingOutcome::FutureEpoch { .. }
-                | AttestationProcessingOutcome::PastEpoch { .. }
-                | AttestationProcessingOutcome::UnknownTargetRoot { .. }
-                | AttestationProcessingOutcome::FinalizedSlot { .. } => {} // ignore the attestation
-                AttestationProcessingOutcome::Invalid { .. }
-                | AttestationProcessingOutcome::EmptyAggregationBitfield { .. }
-                | AttestationProcessingOutcome::AttestsToFutureBlock { .. }
-                | AttestationProcessingOutcome::InvalidSignature
-                | AttestationProcessingOutcome::NoCommitteeForSlotAndIndex { .. }
-                | AttestationProcessingOutcome::BadTargetEpoch { .. } => {
-                    // the peer has sent a bad attestation. Remove them.
-                    self.network.disconnect(peer_id, GoodbyeReason::Fault);
-                }
-            },
-            Err(_) => {
-                // error is logged during the processing therefore no error is logged here
-                trace!(
-                    self.log,
-                    "Erroneous gossip attestation ssz";
-                    "ssz" => format!("0x{}", hex::encode(msg.as_ssz_bytes())),
-                );
-            }
-        };
-        */
-    }
 }
 
 /// Build a `StatusMessage` representing the state of the given `beacon_chain`.

beacon_node/network/src/service.rs

@@ -8,7 +8,7 @@ use crate::{
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use eth2_libp2p::Service as LibP2PService;
 use eth2_libp2p::{rpc::RPCRequest, Enr, Libp2pEvent, MessageId, NetworkGlobals, PeerId, Swarm};
-use eth2_libp2p::{PubsubMessage, RPCEvent};
+use eth2_libp2p::{PubsubData, PubsubMessage, RPCEvent};
 use futures::prelude::*;
 use futures::Stream;
 use rest_types::ValidatorSubscription;
@@ -55,7 +55,7 @@ pub struct NetworkService<T: BeaconChainTypes> {
 impl<T: BeaconChainTypes> NetworkService<T> {
     pub fn start(
         beacon_chain: Arc<BeaconChain<T>>,
-        config: &mut NetworkConfig,
+        config: &NetworkConfig,
         executor: &TaskExecutor,
         network_log: slog::Logger,
     ) -> error::Result<(
@@ -77,8 +77,8 @@ impl<T: BeaconChainTypes> NetworkService<T> {
 
         let propagation_percentage = config.propagation_percentage;
 
-        // set the local enr_fork_id
-        config.enr_fork_id = beacon_chain
+        // build the current enr_fork_id for adding to our local ENR
+        let enr_fork_id = beacon_chain
             .enr_fork_id()
             .map_err(|e| format!("Could not get the current ENR fork version: {:?}", e))?;
 
@@ -88,7 +88,8 @@ impl<T: BeaconChainTypes> NetworkService<T> {
             .map_err(|e| format!("Could not get the next fork update duration: {:?}", e))?;
 
         // launch libp2p service
-        let (network_globals, mut libp2p) = LibP2PService::new(config, network_log.clone())?;
+        let (network_globals, mut libp2p) =
+            LibP2PService::new(config, enr_fork_id, network_log.clone())?;
 
         for enr in load_dht::<T::Store, T::EthSpec>(store.clone()) {
             libp2p.swarm.add_enr(enr);
@@ -262,11 +263,26 @@ fn spawn_service<T: BeaconChainTypes>(
         while let Ok(Async::Ready(Some(attestation_service_message))) = service.attestation_service.poll() {
             match attestation_service_message {
                 // TODO: Implement
-                AttServiceMessage::Subscribe(_subnet) => { },
-                AttServiceMessage::Unsubscribe(_subnet) => { },
-                AttServiceMessage::EnrAdd(_subnet) => { },
-                AttServiceMessage::EnrRemove(_subnet) => { },
-                AttServiceMessage::DiscoverPeers(_subnet) => { },
+                AttServiceMessage::Subscribe(subnet_id) => {
+                    service.libp2p.swarm.subscribe_to_subnet(subnet_id);
+                },
+                AttServiceMessage::Unsubscribe(subnet_id) => {
+                    service.libp2p.swarm.subscribe_to_subnet(subnet_id);
+                 },
+                AttServiceMessage::EnrAdd(subnet_id) => {
+                    service.libp2p.swarm.update_enr_subnet(subnet_id, true);
+                },
+                AttServiceMessage::EnrRemove(subnet_id) => {
+                    service.libp2p.swarm.update_enr_subnet(subnet_id, false);
+                },
+                AttServiceMessage::DiscoverPeers(subnet_id) => {
+                    service.libp2p.swarm.peers_request(subnet_id);
+                },
+                AttServiceMessage::Propagate(source, message_id) => {
+                            service.libp2p
+                                .swarm
+                                .propagate_message(&source, message_id);
+                    }
             }
         }
 
@@ -276,8 +292,6 @@ fn spawn_service<T: BeaconChainTypes>(
             match service.libp2p.poll() {
                 Ok(Async::Ready(Some(event))) => match event {
                     Libp2pEvent::RPC(peer_id, rpc_event) => {
-                        // trace!(log, "Received RPC"; "rpc" => format!("{}", rpc_event));
-
                         // if we received a Goodbye message, drop and ban the peer
                         if let RPCEvent::Request(_, RPCRequest::Goodbye(_)) = rpc_event {
                             peers_to_ban.push(peer_id.clone());
@@ -304,9 +318,24 @@ fn spawn_service<T: BeaconChainTypes>(
                         message,
                         ..
                     } => {
-                       service.router_send
-                            .try_send(RouterMessage::PubsubMessage(id, source, message))
-                            .map_err(|_| { debug!(log, "Failed to send pubsub message to router");})?;
+
+                        match message.data {
+                            // attestation information gets processed in the attestation service
+                            PubsubData::AggregateAndProofAttestation(signed_aggregate_and_proof) => {
+                                service.attestation_service.handle_aggregate_attestation(id, source, *signed_aggregate_and_proof);
+                            },
+                            PubsubData::Attestation(subnet_and_attestation) => {
+                                let subnet = subnet_and_attestation.0;
+                                let attestation = subnet_and_attestation.1;
+                                service.attestation_service.handle_unaggregated_attestation(id, source, subnet, attestation);
+                            }
+                            _ => {
+                                // all else is sent to the router
+                           service.router_send
+                                .try_send(RouterMessage::PubsubMessage(id, source, message))
+                                .map_err(|_| { debug!(log, "Failed to send pubsub message to router");})?;
+                            }
+                        }
                     }
                     Libp2pEvent::PeerSubscribed(_, _) => {}
                 },