Altair networking (#2300)

## Issue Addressed

Resolves #2278 

## Proposed Changes

Implements the networking components for the Altair hard fork https://github.com/ethereum/eth2.0-specs/blob/dev/specs/altair/p2p-interface.md

## Additional Info

This PR acts as the base branch for networking changes and tracks https://github.com/sigp/lighthouse/pull/2279 . Changes to gossip, rpc and discovery can be separate PRs to be merged here for ease of review.

Co-authored-by: realbigsean <seananderson33@gmail.com>

beacon_node/network/src/subnet_service/attestation_subnets.rs

@@ -0,0 +1,645 @@
+//! This service keeps track of which shard subnet the beacon node should be subscribed to at any
+//! given time. It schedules subscriptions to shard subnets, requests peer discoveries and
+//! determines whether attestations should be aggregated and/or passed to the beacon node.
+
+use super::SubnetServiceMessage;
+use std::collections::{HashMap, HashSet, VecDeque};
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::{Duration, Instant};
+
+use futures::prelude::*;
+use rand::seq::SliceRandom;
+use slog::{debug, error, o, trace, warn};
+
+use beacon_chain::{BeaconChain, BeaconChainTypes};
+use eth2_libp2p::{NetworkConfig, Subnet, SubnetDiscovery};
+use hashset_delay::HashSetDelay;
+use slot_clock::SlotClock;
+use types::{Attestation, EthSpec, Slot, SubnetId, ValidatorSubscription};
+
+use crate::metrics;
+
+/// 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.
+/// Subnet discovery query takes atmost 30 secs, 2 slots take 24s.
+const MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD: u64 = 2;
+/// The time (in slots) before a last seen validator is considered absent and we unsubscribe from the random
+/// gossip topics that we subscribed to due to the validator connection.
+const LAST_SEEN_VALIDATOR_TIMEOUT: u32 = 150;
+/// The fraction of a slot that we subscribe to a subnet before the required slot.
+///
+/// Note: The time is calculated as `time = seconds_per_slot / ADVANCE_SUBSCRIPTION_TIME`.
+const ADVANCE_SUBSCRIBE_TIME: u32 = 3;
+/// The default number of slots before items in hash delay sets used by this class should expire.
+///  36s at 12s slot time
+const DEFAULT_EXPIRATION_TIMEOUT: u32 = 3;
+
+/// A particular subnet at a given slot.
+#[derive(PartialEq, Eq, Hash, Clone, Debug)]
+pub struct ExactSubnet {
+    /// The `SubnetId` associated with this subnet.
+    pub subnet_id: SubnetId,
+    /// The `Slot` associated with this subnet.
+    pub slot: Slot,
+}
+
+pub struct AttestationService<T: BeaconChainTypes> {
+    /// Queued events to return to the driving service.
+    events: VecDeque<SubnetServiceMessage>,
+
+    /// A reference to the beacon chain to process received attestations.
+    pub(crate) beacon_chain: Arc<BeaconChain<T>>,
+
+    /// The collection of currently subscribed random subnets mapped to their expiry deadline.
+    pub(crate) random_subnets: HashSetDelay<SubnetId>,
+
+    /// The collection of all currently subscribed subnets (long-lived **and** short-lived).
+    subscriptions: HashSet<SubnetId>,
+
+    /// A collection of timeouts for when to unsubscribe from a shard subnet.
+    unsubscriptions: HashSetDelay<ExactSubnet>,
+
+    /// 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
+    /// our ENR.
+    /// This is a set of validator indices.
+    known_validators: HashSetDelay<u64>,
+
+    /// The waker for the current thread.
+    waker: Option<std::task::Waker>,
+
+    /// The discovery mechanism of lighthouse is disabled.
+    discovery_disabled: bool,
+
+    /// We are always subscribed to all subnets.
+    subscribe_all_subnets: bool,
+
+    /// We process and aggregate all attestations on subscribed subnets.
+    import_all_attestations: bool,
+
+    /// The logger for the attestation service.
+    log: slog::Logger,
+}
+
+impl<T: BeaconChainTypes> AttestationService<T> {
+    /* Public functions */
+
+    pub fn new(
+        beacon_chain: Arc<BeaconChain<T>>,
+        config: &NetworkConfig,
+        log: &slog::Logger,
+    ) -> Self {
+        let log = log.new(o!("service" => "attestation_service"));
+
+        // calculate the random subnet duration from the spec constants
+        let spec = &beacon_chain.spec;
+        let slot_duration = beacon_chain.slot_clock.slot_duration();
+        let random_subnet_duration_millis = spec
+            .epochs_per_random_subnet_subscription
+            .saturating_mul(T::EthSpec::slots_per_epoch())
+            .saturating_mul(slot_duration.as_millis() as u64);
+
+        // Panics on overflow. Ensure LAST_SEEN_VALIDATOR_TIMEOUT is not too large.
+        let last_seen_val_timeout = slot_duration
+            .checked_mul(LAST_SEEN_VALIDATOR_TIMEOUT)
+            .expect("LAST_SEEN_VALIDATOR_TIMEOUT must not be ridiculously large");
+        let default_timeout = slot_duration
+            .checked_mul(DEFAULT_EXPIRATION_TIMEOUT)
+            .expect("DEFAULT_EXPIRATION_TIMEOUT must not be ridiculoustly large");
+
+        AttestationService {
+            events: VecDeque::with_capacity(10),
+            beacon_chain,
+            random_subnets: HashSetDelay::new(Duration::from_millis(random_subnet_duration_millis)),
+            subscriptions: HashSet::new(),
+            unsubscriptions: HashSetDelay::new(default_timeout),
+            aggregate_validators_on_subnet: HashSetDelay::new(default_timeout),
+            known_validators: HashSetDelay::new(last_seen_val_timeout),
+            waker: None,
+            subscribe_all_subnets: config.subscribe_all_subnets,
+            import_all_attestations: config.import_all_attestations,
+            discovery_disabled: config.disable_discovery,
+            log,
+        }
+    }
+
+    /// Return count of all currently subscribed subnets (long-lived **and** short-lived).
+    #[cfg(test)]
+    pub fn subscription_count(&self) -> usize {
+        if self.subscribe_all_subnets {
+            self.beacon_chain.spec.attestation_subnet_count as usize
+        } else {
+            self.subscriptions.len()
+        }
+    }
+
+    /// Processes a list of validator subscriptions.
+    ///
+    /// This will:
+    /// - Register new validators as being known.
+    /// - Subscribe to the required number of random subnets.
+    /// - Update the local ENR for new random subnets due to seeing new validators.
+    /// - Search for peers for required subnets.
+    /// - Request subscriptions for subnets on specific slots when required.
+    /// - Build the timeouts for each of these events.
+    ///
+    /// This returns a result simply for the ergonomics of using ?. The result can be
+    /// safely dropped.
+    pub fn validator_subscriptions(
+        &mut self,
+        subscriptions: Vec<ValidatorSubscription>,
+    ) -> Result<(), String> {
+        // Maps each subnet_id subscription to it's highest slot
+        let mut subnets_to_discover: HashMap<SubnetId, Slot> = HashMap::new();
+        for subscription in subscriptions {
+            metrics::inc_counter(&metrics::SUBNET_SUBSCRIPTION_REQUESTS);
+            //NOTE: We assume all subscriptions have been verified before reaching this service
+
+            // Registers the validator with the attestation service.
+            // This will subscribe to long-lived random subnets if required.
+            trace!(self.log,
+                "Validator subscription";
+                "subscription" => ?subscription,
+            );
+            self.add_known_validator(subscription.validator_index);
+
+            let subnet_id = match SubnetId::compute_subnet::<T::EthSpec>(
+                subscription.slot,
+                subscription.attestation_committee_index,
+                subscription.committee_count_at_slot,
+                &self.beacon_chain.spec,
+            ) {
+                Ok(subnet_id) => subnet_id,
+                Err(e) => {
+                    warn!(self.log,
+                        "Failed to compute subnet id for validator subscription";
+                        "error" => ?e,
+                        "validator_index" => subscription.validator_index
+                    );
+                    continue;
+                }
+            };
+            // Ensure each subnet_id inserted into the map has the highest slot as it's value.
+            // Higher slot corresponds to higher min_ttl in the `SubnetDiscovery` entry.
+            if let Some(slot) = subnets_to_discover.get(&subnet_id) {
+                if subscription.slot > *slot {
+                    subnets_to_discover.insert(subnet_id, subscription.slot);
+                }
+            } else if !self.discovery_disabled {
+                subnets_to_discover.insert(subnet_id, subscription.slot);
+            }
+
+            let exact_subnet = ExactSubnet {
+                subnet_id,
+                slot: subscription.slot,
+            };
+
+            // Determine if the validator is an aggregator. If so, we subscribe to the subnet and
+            // if successful add the validator to a mapping of known aggregators for that exact
+            // subnet.
+
+            if subscription.is_aggregator {
+                metrics::inc_counter(&metrics::SUBNET_SUBSCRIPTION_AGGREGATOR_REQUESTS);
+                // set the subscription timer to subscribe to the next subnet if required
+                if let Err(e) = self.subscribe_to_subnet(exact_subnet.clone()) {
+                    warn!(self.log,
+                        "Subscription to subnet error";
+                        "error" => e,
+                        "validator_index" => subscription.validator_index,
+                    );
+                } else {
+                    trace!(self.log,
+                        "Subscribed to subnet for aggregator duties";
+                        "exact_subnet" => ?exact_subnet,
+                        "validator_index" => subscription.validator_index
+                    );
+                }
+            }
+        }
+
+        // If the discovery mechanism isn't disabled, attempt to set up a peer discovery for the
+        // required subnets.
+        if !self.discovery_disabled {
+            if let Err(e) = self.discover_peers_request(
+                subnets_to_discover
+                    .into_iter()
+                    .map(|(subnet_id, slot)| ExactSubnet { subnet_id, slot }),
+            ) {
+                warn!(self.log, "Discovery lookup request error"; "error" => e);
+            };
+        }
+
+        // pre-emptively wake the thread to check for new events
+        if let Some(waker) = &self.waker {
+            waker.wake_by_ref();
+        }
+        Ok(())
+    }
+
+    /// Checks if we have subscribed aggregate validators for the subnet. If not, checks the gossip
+    /// verification, re-propagates and returns false.
+    pub fn should_process_attestation(
+        &self,
+        subnet: SubnetId,
+        attestation: &Attestation<T::EthSpec>,
+    ) -> bool {
+        if self.import_all_attestations {
+            return true;
+        }
+
+        let exact_subnet = ExactSubnet {
+            subnet_id: subnet,
+            slot: attestation.data.slot,
+        };
+        self.aggregate_validators_on_subnet.contains(&exact_subnet)
+    }
+
+    /* Internal private functions */
+
+    /// Checks if there are currently queued discovery requests and the time required to make the
+    /// request.
+    ///
+    /// If there is sufficient time, queues a peer discovery request for all the required subnets.
+    fn discover_peers_request(
+        &mut self,
+        exact_subnets: impl Iterator<Item = ExactSubnet>,
+    ) -> Result<(), &'static str> {
+        let current_slot = self
+            .beacon_chain
+            .slot_clock
+            .now()
+            .ok_or("Could not get the current slot")?;
+
+        let discovery_subnets: Vec<SubnetDiscovery> = exact_subnets
+            .filter_map(|exact_subnet| {
+                // check if there is enough time to perform a discovery lookup
+                if exact_subnet.slot
+                    >= current_slot.saturating_add(MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD)
+                {
+                    // if the slot is more than epoch away, add an event to start looking for peers
+                    // add one slot to ensure we keep the peer for the subscription slot
+                    let min_ttl = self
+                        .beacon_chain
+                        .slot_clock
+                        .duration_to_slot(exact_subnet.slot + 1)
+                        .map(|duration| std::time::Instant::now() + duration);
+                    Some(SubnetDiscovery {
+                        subnet: Subnet::Attestation(exact_subnet.subnet_id),
+                        min_ttl,
+                    })
+                } else {
+                    // We may want to check the global PeerInfo to see estimated timeouts for each
+                    // peer before they can be removed.
+                    warn!(self.log,
+                        "Not enough time for a discovery search";
+                        "subnet_id" => ?exact_subnet
+                    );
+                    None
+                }
+            })
+            .collect();
+
+        if !discovery_subnets.is_empty() {
+            self.events
+                .push_back(SubnetServiceMessage::DiscoverPeers(discovery_subnets));
+        }
+        Ok(())
+    }
+
+    /// Checks the current random subnets and subscriptions to determine if a new subscription for this
+    /// subnet is required for the given slot.
+    ///
+    /// If required, adds a subscription event and an associated unsubscription event.
+    fn subscribe_to_subnet(&mut self, exact_subnet: ExactSubnet) -> Result<(), &'static str> {
+        // initialise timing variables
+        let current_slot = self
+            .beacon_chain
+            .slot_clock
+            .now()
+            .ok_or("Could not get the current slot")?;
+
+        // Calculate the duration to the unsubscription event.
+        // There are two main cases. Attempting to subscribe to the current slot and all others.
+        let expected_end_subscription_duration = if current_slot >= exact_subnet.slot {
+            self.beacon_chain
+                .slot_clock
+                .duration_to_next_slot()
+                .ok_or("Unable to determine duration to next slot")?
+        } else {
+            let slot_duration = self.beacon_chain.slot_clock.slot_duration();
+
+            // the duration until we no longer need this subscription. We assume a single slot is
+            // sufficient.
+            self.beacon_chain
+                .slot_clock
+                .duration_to_slot(exact_subnet.slot)
+                .ok_or("Unable to determine duration to subscription slot")?
+                + slot_duration
+        };
+
+        // 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
+        // long-lived random subnet can be unsubscribed at any time when a validator becomes
+        // in-active. This case is checked on the subscription event (see `handle_subscriptions`).
+
+        // Return if we already have a subscription for this subnet_id and slot
+        if self.unsubscriptions.contains(&exact_subnet) || self.subscribe_all_subnets {
+            return Ok(());
+        }
+
+        // We are not currently subscribed and have no waiting subscription, create one
+        self.handle_subscriptions(exact_subnet.clone());
+
+        // if there is an unsubscription event for the slot prior, we remove it to prevent
+        // unsubscriptions immediately after the subscription. We also want to minimize
+        // subscription churn and maintain a consecutive subnet subscriptions.
+        self.unsubscriptions.retain(|subnet| {
+            !(subnet.subnet_id == exact_subnet.subnet_id && subnet.slot <= exact_subnet.slot)
+        });
+        // add an unsubscription event to remove ourselves from the subnet once completed
+        self.unsubscriptions
+            .insert_at(exact_subnet, expected_end_subscription_duration);
+        Ok(())
+    }
+
+    /// Updates the `known_validators` mapping and subscribes to a set of random subnets if required.
+    ///
+    /// This also updates the ENR to indicate our long-lived subscription to the subnet
+    fn add_known_validator(&mut self, validator_index: u64) {
+        if self.known_validators.get(&validator_index).is_none() && !self.subscribe_all_subnets {
+            // New validator has subscribed
+            // Subscribe to random topics and update the ENR if needed.
+
+            let spec = &self.beacon_chain.spec;
+
+            if self.random_subnets.len() < spec.attestation_subnet_count as usize {
+                // Still room for subscriptions
+                self.subscribe_to_random_subnets(
+                    self.beacon_chain.spec.random_subnets_per_validator as usize,
+                );
+            }
+        }
+        // add the new validator or update the current timeout for a known validator
+        self.known_validators.insert(validator_index);
+    }
+
+    /// Subscribe to long-lived random subnets and update the local ENR bitfield.
+    fn subscribe_to_random_subnets(&mut self, no_subnets_to_subscribe: usize) {
+        let subnet_count = self.beacon_chain.spec.attestation_subnet_count;
+
+        // Build a list of random subnets that we are not currently subscribed to.
+        let available_subnets = (0..subnet_count)
+            .map(SubnetId::new)
+            .filter(|subnet_id| self.random_subnets.get(subnet_id).is_none())
+            .collect::<Vec<_>>();
+
+        let to_subscribe_subnets = {
+            if available_subnets.len() < no_subnets_to_subscribe {
+                debug!(self.log, "Reached maximum random subnet subscriptions");
+                available_subnets
+            } else {
+                // select a random sample of available subnets
+                available_subnets
+                    .choose_multiple(&mut rand::thread_rng(), no_subnets_to_subscribe)
+                    .cloned()
+                    .collect::<Vec<_>>()
+            }
+        };
+
+        for subnet_id in to_subscribe_subnets {
+            // remove this subnet from any immediate un-subscription events
+            self.unsubscriptions
+                .retain(|exact_subnet| exact_subnet.subnet_id != subnet_id);
+
+            // insert a new random subnet
+            self.random_subnets.insert(subnet_id);
+
+            // send discovery request
+            // Note: it's wasteful to send a DiscoverPeers request if we already have peers for this subnet.
+            // However, subscribing to random subnets ideally shouldn't happen very often (once in ~27 hours) and
+            // this makes it easier to deterministically test the attestations service.
+            self.events
+                .push_back(SubnetServiceMessage::DiscoverPeers(vec![SubnetDiscovery {
+                    subnet: Subnet::Attestation(subnet_id),
+                    min_ttl: None,
+                }]));
+
+            // if we are not already subscribed, then subscribe
+            if !self.subscriptions.contains(&subnet_id) {
+                self.subscriptions.insert(subnet_id);
+                debug!(self.log, "Subscribing to random subnet"; "subnet_id" => ?subnet_id);
+                self.events
+                    .push_back(SubnetServiceMessage::Subscribe(Subnet::Attestation(
+                        subnet_id,
+                    )));
+            }
+
+            // add the subnet to the ENR bitfield
+            self.events
+                .push_back(SubnetServiceMessage::EnrAdd(Subnet::Attestation(subnet_id)));
+        }
+    }
+
+    /* A collection of functions that handle the various timeouts */
+
+    /// A queued subscription is ready.
+    ///
+    /// We add subscriptions events even if we are already subscribed to a random subnet (as these
+    /// can be unsubscribed at any time by inactive validators). If we are
+    /// still subscribed at the time the event fires, we don't re-subscribe.
+    fn handle_subscriptions(&mut self, exact_subnet: ExactSubnet) {
+        // Check if the subnet currently exists as a long-lasting random subnet
+        if let Some(expiry) = self.random_subnets.get(&exact_subnet.subnet_id) {
+            // we are subscribed via a random subnet, if this is to expire during the time we need
+            // to be subscribed, just extend the expiry
+            let slot_duration = self.beacon_chain.slot_clock.slot_duration();
+            let advance_subscription_duration = slot_duration
+                .checked_div(ADVANCE_SUBSCRIBE_TIME)
+                .expect("ADVANCE_SUBSCRIPTION_TIME cannot be too large");
+            // we require the subnet subscription for at least a slot on top of the initial
+            // subscription time
+            let expected_end_subscription_duration = slot_duration + advance_subscription_duration;
+
+            if expiry < &(Instant::now() + expected_end_subscription_duration) {
+                self.random_subnets
+                    .update_timeout(&exact_subnet.subnet_id, expected_end_subscription_duration);
+            }
+        } else {
+            // we are also not un-subscribing from a subnet if the next slot requires us to be
+            // subscribed. Therefore there could be the case that we are already still subscribed
+            // to the required subnet. In which case we do not issue another subscription request.
+            if !self.subscriptions.contains(&exact_subnet.subnet_id) {
+                // we are not already subscribed
+                debug!(self.log, "Subscribing to subnet"; "subnet" => *exact_subnet.subnet_id, "target_slot" => exact_subnet.slot.as_u64());
+                self.subscriptions.insert(exact_subnet.subnet_id);
+                self.events
+                    .push_back(SubnetServiceMessage::Subscribe(Subnet::Attestation(
+                        exact_subnet.subnet_id,
+                    )));
+            }
+        }
+    }
+
+    /// A queued unsubscription is ready.
+    ///
+    /// Unsubscription events are added, even if we are subscribed to long-lived random subnets. If
+    /// a random subnet is present, we do not unsubscribe from it.
+    fn handle_unsubscriptions(&mut self, exact_subnet: ExactSubnet) {
+        // Check if the subnet currently exists as a long-lasting random subnet
+        if self.random_subnets.contains(&exact_subnet.subnet_id) {
+            return;
+        }
+
+        debug!(self.log, "Unsubscribing from subnet"; "subnet" => *exact_subnet.subnet_id, "processed_slot" => exact_subnet.slot.as_u64());
+
+        self.subscriptions.remove(&exact_subnet.subnet_id);
+        self.events
+            .push_back(SubnetServiceMessage::Unsubscribe(Subnet::Attestation(
+                exact_subnet.subnet_id,
+            )));
+    }
+
+    /// A random subnet has expired.
+    ///
+    /// This function selects a new subnet to join, or extends the expiry if there are no more
+    /// available subnets to choose from.
+    fn handle_random_subnet_expiry(&mut self, subnet_id: SubnetId) {
+        let subnet_count = self.beacon_chain.spec.attestation_subnet_count;
+        if self.random_subnets.len() == (subnet_count - 1) as usize {
+            // We are at capacity, simply increase the timeout of the current subnet
+            self.random_subnets.insert(subnet_id);
+            return;
+        }
+        // If there are no unsubscription events for `subnet_id`, we unsubscribe immediately.
+        if !self
+            .unsubscriptions
+            .keys()
+            .any(|s| s.subnet_id == subnet_id)
+        {
+            // we are not at capacity, unsubscribe from the current subnet.
+            debug!(self.log, "Unsubscribing from random subnet"; "subnet_id" => *subnet_id);
+            self.events
+                .push_back(SubnetServiceMessage::Unsubscribe(Subnet::Attestation(
+                    subnet_id,
+                )));
+        }
+
+        // Remove the ENR bitfield bit and choose a new random on from the available subnets
+        self.events
+            .push_back(SubnetServiceMessage::EnrRemove(Subnet::Attestation(
+                subnet_id,
+            )));
+        // Subscribe to a new random subnet
+        self.subscribe_to_random_subnets(1);
+    }
+
+    /// A known validator has not sent a subscription in a while. They are considered offline and the
+    /// beacon node no longer needs to be subscribed to the allocated random subnets.
+    ///
+    /// We don't keep track of a specific validator to random subnet, rather the ratio of active
+    /// validators to random subnets. So when a validator goes offline, we can simply remove the
+    /// allocated amount of random subnets.
+    fn handle_known_validator_expiry(&mut self) {
+        let spec = &self.beacon_chain.spec;
+        let subnet_count = spec.attestation_subnet_count;
+        let random_subnets_per_validator = spec.random_subnets_per_validator;
+        if self.known_validators.len() as u64 * random_subnets_per_validator >= subnet_count {
+            // have too many validators, ignore
+            return;
+        }
+
+        let subscribed_subnets = self.random_subnets.keys().cloned().collect::<Vec<_>>();
+        let to_remove_subnets = subscribed_subnets.choose_multiple(
+            &mut rand::thread_rng(),
+            random_subnets_per_validator as usize,
+        );
+
+        for subnet_id in to_remove_subnets {
+            // If there are no unsubscription events for `subnet_id`, we unsubscribe immediately.
+            if !self
+                .unsubscriptions
+                .keys()
+                .any(|s| s.subnet_id == *subnet_id)
+            {
+                self.events
+                    .push_back(SubnetServiceMessage::Unsubscribe(Subnet::Attestation(
+                        *subnet_id,
+                    )));
+            }
+            // as the long lasting subnet subscription is being removed, remove the subnet_id from
+            // the ENR bitfield
+            self.events
+                .push_back(SubnetServiceMessage::EnrRemove(Subnet::Attestation(
+                    *subnet_id,
+                )));
+            self.random_subnets.remove(subnet_id);
+        }
+    }
+}
+
+impl<T: BeaconChainTypes> Stream for AttestationService<T> {
+    type Item = SubnetServiceMessage;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // update the waker if needed
+        if let Some(waker) = &self.waker {
+            if waker.will_wake(cx.waker()) {
+                self.waker = Some(cx.waker().clone());
+            }
+        } else {
+            self.waker = Some(cx.waker().clone());
+        }
+
+        // process any un-subscription events
+        match self.unsubscriptions.poll_next_unpin(cx) {
+            Poll::Ready(Some(Ok(exact_subnet))) => self.handle_unsubscriptions(exact_subnet),
+            Poll::Ready(Some(Err(e))) => {
+                error!(self.log, "Failed to check for subnet unsubscription times"; "error"=> e);
+            }
+            Poll::Ready(None) | Poll::Pending => {}
+        }
+
+        // process any random subnet expiries
+        match self.random_subnets.poll_next_unpin(cx) {
+            Poll::Ready(Some(Ok(subnet))) => self.handle_random_subnet_expiry(subnet),
+            Poll::Ready(Some(Err(e))) => {
+                error!(self.log, "Failed to check for random subnet cycles"; "error"=> e);
+            }
+            Poll::Ready(None) | Poll::Pending => {}
+        }
+
+        // process any known validator expiries
+        match self.known_validators.poll_next_unpin(cx) {
+            Poll::Ready(Some(Ok(_validator_index))) => {
+                let _ = self.handle_known_validator_expiry();
+            }
+            Poll::Ready(Some(Err(e))) => {
+                error!(self.log, "Failed to check for random subnet cycles"; "error"=> e);
+            }
+            Poll::Ready(None) | Poll::Pending => {}
+        }
+        // poll to remove entries on expiration, no need to act on expiration events
+        if let Poll::Ready(Some(Err(e))) = self.aggregate_validators_on_subnet.poll_next_unpin(cx) {
+            error!(self.log, "Failed to check for aggregate validator on subnet expirations"; "error"=> e);
+        }
+
+        // process any generated events
+        if let Some(event) = self.events.pop_front() {
+            return Poll::Ready(Some(event));
+        }
+
+        Poll::Pending
+    }
+}

beacon_node/network/src/subnet_service/mod.rs

@@ -0,0 +1,50 @@
+pub mod attestation_subnets;
+pub mod sync_subnets;
+
+use eth2_libp2p::{Subnet, SubnetDiscovery};
+
+pub use attestation_subnets::AttestationService;
+pub use sync_subnets::SyncCommitteeService;
+
+#[cfg(test)]
+mod tests;
+
+#[derive(Debug, Clone)]
+pub enum SubnetServiceMessage {
+    /// Subscribe to the specified subnet id.
+    Subscribe(Subnet),
+    /// Unsubscribe to the specified subnet id.
+    Unsubscribe(Subnet),
+    /// Add the `SubnetId` to the ENR bitfield.
+    EnrAdd(Subnet),
+    /// Remove the `SubnetId` from the ENR bitfield.
+    EnrRemove(Subnet),
+    /// Discover peers for a list of `SubnetDiscovery`.
+    DiscoverPeers(Vec<SubnetDiscovery>),
+}
+
+/// Note: This `PartialEq` impl is for use only in tests.
+/// The `DiscoverPeers` comparison is good enough for testing only.
+#[cfg(test)]
+impl PartialEq for SubnetServiceMessage {
+    fn eq(&self, other: &SubnetServiceMessage) -> bool {
+        match (self, other) {
+            (SubnetServiceMessage::Subscribe(a), SubnetServiceMessage::Subscribe(b)) => a == b,
+            (SubnetServiceMessage::Unsubscribe(a), SubnetServiceMessage::Unsubscribe(b)) => a == b,
+            (SubnetServiceMessage::EnrAdd(a), SubnetServiceMessage::EnrAdd(b)) => a == b,
+            (SubnetServiceMessage::EnrRemove(a), SubnetServiceMessage::EnrRemove(b)) => a == b,
+            (SubnetServiceMessage::DiscoverPeers(a), SubnetServiceMessage::DiscoverPeers(b)) => {
+                if a.len() != b.len() {
+                    return false;
+                }
+                for i in 0..a.len() {
+                    if a[i].subnet != b[i].subnet || a[i].min_ttl != b[i].min_ttl {
+                        return false;
+                    }
+                }
+                true
+            }
+            _ => false,
+        }
+    }
+}

beacon_node/network/src/subnet_service/sync_subnets.rs

@@ -0,0 +1,350 @@
+//! This service keeps track of which sync committee subnet the beacon node should be subscribed to at any
+//! given time. It schedules subscriptions to sync committee subnets and requests peer discoveries.
+
+use std::collections::{hash_map::Entry, HashMap, VecDeque};
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+use futures::prelude::*;
+use slog::{debug, error, o, trace, warn};
+
+use super::SubnetServiceMessage;
+use beacon_chain::{BeaconChain, BeaconChainTypes};
+use eth2_libp2p::{NetworkConfig, Subnet, SubnetDiscovery};
+use hashset_delay::HashSetDelay;
+use slot_clock::SlotClock;
+use types::{Epoch, EthSpec, SyncCommitteeSubscription, SyncSubnetId};
+
+use crate::metrics;
+
+/// 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.
+/// Subnet discovery query takes atmost 30 secs, 2 slots take 24s.
+const MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD: u64 = 2;
+
+/// A particular subnet at a given slot.
+#[derive(PartialEq, Eq, Hash, Clone, Debug)]
+pub struct ExactSubnet {
+    /// The `SyncSubnetId` associated with this subnet.
+    pub subnet_id: SyncSubnetId,
+    /// The epoch until which we need to stay subscribed to the subnet.
+    pub until_epoch: Epoch,
+}
+pub struct SyncCommitteeService<T: BeaconChainTypes> {
+    /// Queued events to return to the driving service.
+    events: VecDeque<SubnetServiceMessage>,
+
+    /// A reference to the beacon chain to process received attestations.
+    pub(crate) beacon_chain: Arc<BeaconChain<T>>,
+
+    /// The collection of all currently subscribed subnets.
+    subscriptions: HashMap<SyncSubnetId, Epoch>,
+
+    /// A collection of timeouts for when to unsubscribe from a subnet.
+    unsubscriptions: HashSetDelay<SyncSubnetId>,
+
+    /// The waker for the current thread.
+    waker: Option<std::task::Waker>,
+
+    /// The discovery mechanism of lighthouse is disabled.
+    discovery_disabled: bool,
+
+    /// We are always subscribed to all subnets.
+    subscribe_all_subnets: bool,
+
+    /// The logger for the attestation service.
+    log: slog::Logger,
+}
+
+impl<T: BeaconChainTypes> SyncCommitteeService<T> {
+    /* Public functions */
+
+    pub fn new(
+        beacon_chain: Arc<BeaconChain<T>>,
+        config: &NetworkConfig,
+        log: &slog::Logger,
+    ) -> Self {
+        let log = log.new(o!("service" => "sync_committee_service"));
+
+        let spec = &beacon_chain.spec;
+        let epoch_duration_secs =
+            beacon_chain.slot_clock.slot_duration().as_secs() * T::EthSpec::slots_per_epoch();
+        let default_timeout =
+            epoch_duration_secs.saturating_mul(spec.epochs_per_sync_committee_period.as_u64());
+
+        SyncCommitteeService {
+            events: VecDeque::with_capacity(10),
+            beacon_chain,
+            subscriptions: HashMap::new(),
+            unsubscriptions: HashSetDelay::new(Duration::from_secs(default_timeout)),
+            waker: None,
+            subscribe_all_subnets: config.subscribe_all_subnets,
+            discovery_disabled: config.disable_discovery,
+            log,
+        }
+    }
+
+    /// Return count of all currently subscribed subnets.
+    #[cfg(test)]
+    pub fn subscription_count(&self) -> usize {
+        use types::consts::altair::SYNC_COMMITTEE_SUBNET_COUNT;
+        if self.subscribe_all_subnets {
+            SYNC_COMMITTEE_SUBNET_COUNT as usize
+        } else {
+            self.subscriptions.len()
+        }
+    }
+
+    /// Processes a list of sync committee subscriptions.
+    ///
+    /// This will:
+    /// - Search for peers for required subnets.
+    /// - Request subscriptions required subnets.
+    /// - Build the timeouts for each of these events.
+    ///
+    /// This returns a result simply for the ergonomics of using ?. The result can be
+    /// safely dropped.
+    pub fn validator_subscriptions(
+        &mut self,
+        subscriptions: Vec<SyncCommitteeSubscription>,
+    ) -> Result<(), String> {
+        let mut subnets_to_discover = Vec::new();
+        for subscription in subscriptions {
+            metrics::inc_counter(&metrics::SYNC_COMMITTEE_SUBSCRIPTION_REQUESTS);
+            //NOTE: We assume all subscriptions have been verified before reaching this service
+
+            // Registers the validator with the subnet  service.
+            // This will subscribe to long-lived random subnets if required.
+            trace!(self.log,
+                "Sync committee subscription";
+                "subscription" => ?subscription,
+            );
+
+            let subnet_ids = match SyncSubnetId::compute_subnets_for_sync_committee::<T::EthSpec>(
+                &subscription.sync_committee_indices,
+            ) {
+                Ok(subnet_ids) => subnet_ids,
+                Err(e) => {
+                    warn!(self.log,
+                        "Failed to compute subnet id for sync committee subscription";
+                        "error" => ?e,
+                        "validator_index" => subscription.validator_index
+                    );
+                    continue;
+                }
+            };
+
+            for subnet_id in subnet_ids {
+                let exact_subnet = ExactSubnet {
+                    subnet_id,
+                    until_epoch: subscription.until_epoch,
+                };
+                subnets_to_discover.push(exact_subnet.clone());
+                if let Err(e) = self.subscribe_to_subnet(exact_subnet.clone()) {
+                    warn!(self.log,
+                        "Subscription to sync subnet error";
+                        "error" => e,
+                        "validator_index" => subscription.validator_index,
+                    );
+                } else {
+                    trace!(self.log,
+                        "Subscribed to subnet for sync committee duties";
+                        "exact_subnet" => ?exact_subnet,
+                        "validator_index" => subscription.validator_index
+                    );
+                }
+            }
+        }
+        // If the discovery mechanism isn't disabled, attempt to set up a peer discovery for the
+        // required subnets.
+        if !self.discovery_disabled {
+            if let Err(e) = self.discover_peers_request(subnets_to_discover.iter()) {
+                warn!(self.log, "Discovery lookup request error"; "error" => e);
+            };
+        }
+
+        // pre-emptively wake the thread to check for new events
+        if let Some(waker) = &self.waker {
+            waker.wake_by_ref();
+        }
+        Ok(())
+    }
+
+    /* Internal private functions */
+
+    /// Checks if there are currently queued discovery requests and the time required to make the
+    /// request.
+    ///
+    /// If there is sufficient time, queues a peer discovery request for all the required subnets.
+    fn discover_peers_request<'a>(
+        &mut self,
+        exact_subnets: impl Iterator<Item = &'a ExactSubnet>,
+    ) -> Result<(), &'static str> {
+        let current_slot = self
+            .beacon_chain
+            .slot_clock
+            .now()
+            .ok_or("Could not get the current slot")?;
+
+        let slots_per_epoch = T::EthSpec::slots_per_epoch();
+
+        let discovery_subnets: Vec<SubnetDiscovery> = exact_subnets
+            .filter_map(|exact_subnet| {
+                let until_slot = exact_subnet.until_epoch.end_slot(slots_per_epoch);
+                // check if there is enough time to perform a discovery lookup
+                if until_slot >= current_slot.saturating_add(MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD) {
+                    // if the slot is more than epoch away, add an event to start looking for peers
+                    // add one slot to ensure we keep the peer for the subscription slot
+                    let min_ttl = self
+                        .beacon_chain
+                        .slot_clock
+                        .duration_to_slot(until_slot + 1)
+                        .map(|duration| std::time::Instant::now() + duration);
+                    Some(SubnetDiscovery {
+                        subnet: Subnet::SyncCommittee(exact_subnet.subnet_id),
+                        min_ttl,
+                    })
+                } else {
+                    // We may want to check the global PeerInfo to see estimated timeouts for each
+                    // peer before they can be removed.
+                    warn!(self.log,
+                        "Not enough time for a discovery search";
+                        "subnet_id" => ?exact_subnet
+                    );
+                    None
+                }
+            })
+            .collect();
+
+        if !discovery_subnets.is_empty() {
+            self.events
+                .push_back(SubnetServiceMessage::DiscoverPeers(discovery_subnets));
+        }
+        Ok(())
+    }
+
+    /// Adds a subscription event and an associated unsubscription event if required.
+    fn subscribe_to_subnet(&mut self, exact_subnet: ExactSubnet) -> Result<(), &'static str> {
+        // Return if we have subscribed to all subnets
+        if self.subscribe_all_subnets {
+            return Ok(());
+        }
+
+        // Return if we already have a subscription for exact_subnet
+        if self.subscriptions.get(&exact_subnet.subnet_id) == Some(&exact_subnet.until_epoch) {
+            return Ok(());
+        }
+
+        // Return if we already have subscription set to expire later than the current request.
+        if let Some(until_epoch) = self.subscriptions.get(&exact_subnet.subnet_id) {
+            if *until_epoch >= exact_subnet.until_epoch {
+                return Ok(());
+            }
+        }
+
+        // initialise timing variables
+        let current_slot = self
+            .beacon_chain
+            .slot_clock
+            .now()
+            .ok_or("Could not get the current slot")?;
+
+        let slots_per_epoch = T::EthSpec::slots_per_epoch();
+        let until_slot = exact_subnet.until_epoch.end_slot(slots_per_epoch);
+        // Calculate the duration to the unsubscription event.
+        let expected_end_subscription_duration = if current_slot >= until_slot {
+            warn!(
+                self.log,
+                "Sync committee subscription is past expiration";
+                "current_slot" => current_slot,
+                "exact_subnet" => ?exact_subnet,
+            );
+            return Ok(());
+        } else {
+            let slot_duration = self.beacon_chain.slot_clock.slot_duration();
+
+            // the duration until we no longer need this subscription. We assume a single slot is
+            // sufficient.
+            self.beacon_chain
+                .slot_clock
+                .duration_to_slot(until_slot)
+                .ok_or("Unable to determine duration to unsubscription slot")?
+                + slot_duration
+        };
+
+        if let Entry::Vacant(e) = self.subscriptions.entry(exact_subnet.subnet_id) {
+            // We are not currently subscribed and have no waiting subscription, create one
+            debug!(self.log, "Subscribing to subnet"; "subnet" => *exact_subnet.subnet_id, "until_epoch" => ?exact_subnet.until_epoch);
+            e.insert(exact_subnet.until_epoch);
+            self.events
+                .push_back(SubnetServiceMessage::Subscribe(Subnet::SyncCommittee(
+                    exact_subnet.subnet_id,
+                )));
+
+            // add the subnet to the ENR bitfield
+            self.events
+                .push_back(SubnetServiceMessage::EnrAdd(Subnet::SyncCommittee(
+                    exact_subnet.subnet_id,
+                )));
+
+            // add an unsubscription event to remove ourselves from the subnet once completed
+            self.unsubscriptions
+                .insert_at(exact_subnet.subnet_id, expected_end_subscription_duration);
+        } else {
+            // We are already subscribed, extend the unsubscription duration
+            self.unsubscriptions
+                .update_timeout(&exact_subnet.subnet_id, expected_end_subscription_duration);
+        }
+
+        Ok(())
+    }
+
+    /// A queued unsubscription is ready.
+    fn handle_unsubscriptions(&mut self, subnet_id: SyncSubnetId) {
+        debug!(self.log, "Unsubscribing from subnet"; "subnet" => *subnet_id);
+
+        self.subscriptions.remove(&subnet_id);
+        self.events
+            .push_back(SubnetServiceMessage::Unsubscribe(Subnet::SyncCommittee(
+                subnet_id,
+            )));
+
+        self.events
+            .push_back(SubnetServiceMessage::EnrRemove(Subnet::SyncCommittee(
+                subnet_id,
+            )));
+    }
+}
+
+impl<T: BeaconChainTypes> Stream for SyncCommitteeService<T> {
+    type Item = SubnetServiceMessage;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // update the waker if needed
+        if let Some(waker) = &self.waker {
+            if waker.will_wake(cx.waker()) {
+                self.waker = Some(cx.waker().clone());
+            }
+        } else {
+            self.waker = Some(cx.waker().clone());
+        }
+
+        // process any un-subscription events
+        match self.unsubscriptions.poll_next_unpin(cx) {
+            Poll::Ready(Some(Ok(exact_subnet))) => self.handle_unsubscriptions(exact_subnet),
+            Poll::Ready(Some(Err(e))) => {
+                error!(self.log, "Failed to check for subnet unsubscription times"; "error"=> e);
+            }
+            Poll::Ready(None) | Poll::Pending => {}
+        }
+
+        // process any generated events
+        if let Some(event) = self.events.pop_front() {
+            return Poll::Ready(Some(event));
+        }
+
+        Poll::Pending
+    }
+}

beacon_node/network/src/subnet_service/tests/mod.rs

@@ -0,0 +1,573 @@
+use super::*;
+use beacon_chain::{
+    builder::{BeaconChainBuilder, Witness},
+    eth1_chain::CachingEth1Backend,
+    BeaconChain,
+};
+use eth2_libp2p::NetworkConfig;
+use futures::prelude::*;
+use genesis::{generate_deterministic_keypairs, interop_genesis_state};
+use lazy_static::lazy_static;
+use slog::Logger;
+use sloggers::{null::NullLoggerBuilder, Build};
+use slot_clock::{SlotClock, SystemTimeSlotClock};
+use std::sync::Arc;
+use std::time::{Duration, SystemTime};
+use store::config::StoreConfig;
+use store::{HotColdDB, MemoryStore};
+use types::{
+    CommitteeIndex, Epoch, EthSpec, MainnetEthSpec, Slot, SubnetId, SyncCommitteeSubscription,
+    SyncSubnetId, ValidatorSubscription,
+};
+
+const SLOT_DURATION_MILLIS: u64 = 400;
+
+type TestBeaconChainType = Witness<
+    SystemTimeSlotClock,
+    CachingEth1Backend<MainnetEthSpec>,
+    MainnetEthSpec,
+    MemoryStore<MainnetEthSpec>,
+    MemoryStore<MainnetEthSpec>,
+>;
+
+pub struct TestBeaconChain {
+    chain: Arc<BeaconChain<TestBeaconChainType>>,
+}
+
+impl TestBeaconChain {
+    pub fn new_with_system_clock() -> Self {
+        let spec = MainnetEthSpec::default_spec();
+
+        let keypairs = generate_deterministic_keypairs(1);
+
+        let log = get_logger();
+        let store =
+            HotColdDB::open_ephemeral(StoreConfig::default(), spec.clone(), log.clone()).unwrap();
+
+        let (shutdown_tx, _) = futures::channel::mpsc::channel(1);
+
+        let chain = Arc::new(
+            BeaconChainBuilder::new(MainnetEthSpec)
+                .logger(log.clone())
+                .custom_spec(spec.clone())
+                .store(Arc::new(store))
+                .genesis_state(
+                    interop_genesis_state::<MainnetEthSpec>(&keypairs, 0, &spec)
+                        .expect("should generate interop state"),
+                )
+                .expect("should build state using recent genesis")
+                .dummy_eth1_backend()
+                .expect("should build dummy backend")
+                .slot_clock(SystemTimeSlotClock::new(
+                    Slot::new(0),
+                    Duration::from_secs(recent_genesis_time()),
+                    Duration::from_millis(SLOT_DURATION_MILLIS),
+                ))
+                .shutdown_sender(shutdown_tx)
+                .monitor_validators(true, vec![], log)
+                .build()
+                .expect("should build"),
+        );
+        Self { chain }
+    }
+}
+
+pub fn recent_genesis_time() -> u64 {
+    SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+        .as_secs()
+}
+
+fn get_logger() -> Logger {
+    NullLoggerBuilder.build().expect("logger should build")
+}
+
+lazy_static! {
+    static ref CHAIN: TestBeaconChain = TestBeaconChain::new_with_system_clock();
+}
+
+fn get_attestation_service() -> AttestationService<TestBeaconChainType> {
+    let log = get_logger();
+    let config = NetworkConfig::default();
+
+    let beacon_chain = CHAIN.chain.clone();
+
+    AttestationService::new(beacon_chain, &config, &log)
+}
+
+fn get_sync_committee_service() -> SyncCommitteeService<TestBeaconChainType> {
+    let log = get_logger();
+    let config = NetworkConfig::default();
+
+    let beacon_chain = CHAIN.chain.clone();
+
+    SyncCommitteeService::new(beacon_chain, &config, &log)
+}
+
+// gets a number of events from the subscription service, or returns none if it times out after a number
+// of slots
+async fn get_events<S: Stream<Item = SubnetServiceMessage> + Unpin>(
+    stream: &mut S,
+    num_events: Option<usize>,
+    num_slots_before_timeout: u32,
+) -> Vec<SubnetServiceMessage> {
+    let mut events = Vec::new();
+
+    let collect_stream_fut = async {
+        loop {
+            if let Some(result) = stream.next().await {
+                events.push(result);
+                if let Some(num) = num_events {
+                    if events.len() == num {
+                        return;
+                    }
+                }
+            }
+        }
+    };
+
+    tokio::select! {
+        _ = collect_stream_fut => events,
+        _ = tokio::time::sleep(
+        Duration::from_millis(SLOT_DURATION_MILLIS) * num_slots_before_timeout,
+    ) => events
+    }
+}
+
+mod attestation_service {
+    use super::*;
+
+    fn get_subscription(
+        validator_index: u64,
+        attestation_committee_index: CommitteeIndex,
+        slot: Slot,
+        committee_count_at_slot: u64,
+    ) -> ValidatorSubscription {
+        let is_aggregator = true;
+        ValidatorSubscription {
+            validator_index,
+            attestation_committee_index,
+            slot,
+            committee_count_at_slot,
+            is_aggregator,
+        }
+    }
+
+    fn get_subscriptions(
+        validator_count: u64,
+        slot: Slot,
+        committee_count_at_slot: u64,
+    ) -> Vec<ValidatorSubscription> {
+        (0..validator_count)
+            .map(|validator_index| {
+                get_subscription(
+                    validator_index,
+                    validator_index,
+                    slot,
+                    committee_count_at_slot,
+                )
+            })
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn subscribe_current_slot_wait_for_unsubscribe() {
+        // subscription config
+        let validator_index = 1;
+        let committee_index = 1;
+        // Keep a low subscription slot so that there are no additional subnet discovery events.
+        let subscription_slot = 0;
+        let committee_count = 1;
+
+        // create the attestation service and subscriptions
+        let mut attestation_service = get_attestation_service();
+        let current_slot = attestation_service
+            .beacon_chain
+            .slot_clock
+            .now()
+            .expect("Could not get current slot");
+
+        let subscriptions = vec![get_subscription(
+            validator_index,
+            committee_index,
+            current_slot + Slot::new(subscription_slot),
+            committee_count,
+        )];
+
+        // submit the subscriptions
+        attestation_service
+            .validator_subscriptions(subscriptions)
+            .unwrap();
+
+        // not enough time for peer discovery, just subscribe, unsubscribe
+        let subnet_id = SubnetId::compute_subnet::<MainnetEthSpec>(
+            current_slot + Slot::new(subscription_slot),
+            committee_index,
+            committee_count,
+            &attestation_service.beacon_chain.spec,
+        )
+        .unwrap();
+        let expected = vec![
+            SubnetServiceMessage::Subscribe(Subnet::Attestation(subnet_id)),
+            SubnetServiceMessage::Unsubscribe(Subnet::Attestation(subnet_id)),
+        ];
+
+        // Wait for 1 slot duration to get the unsubscription event
+        let events = get_events(
+            &mut attestation_service,
+            Some(5),
+            (MainnetEthSpec::slots_per_epoch() * 3) as u32,
+        )
+        .await;
+        matches::assert_matches!(
+            events[..3],
+            [
+                SubnetServiceMessage::DiscoverPeers(_),
+                SubnetServiceMessage::Subscribe(_any1),
+                SubnetServiceMessage::EnrAdd(_any3)
+            ]
+        );
+
+        // If the long lived and short lived subnets are the same, there should be no more events
+        // as we don't resubscribe already subscribed subnets.
+        if !attestation_service.random_subnets.contains(&subnet_id) {
+            assert_eq!(expected[..], events[3..]);
+        }
+        // Should be subscribed to only 1 long lived subnet after unsubscription.
+        assert_eq!(attestation_service.subscription_count(), 1);
+    }
+
+    /// Test to verify that we are not unsubscribing to a subnet before a required subscription.
+    #[tokio::test]
+    async fn test_same_subnet_unsubscription() {
+        // subscription config
+        let validator_index = 1;
+        let committee_count = 1;
+
+        // Makes 2 validator subscriptions to the same subnet but at different slots.
+        // There should be just 1 unsubscription event for the later slot subscription (subscription_slot2).
+        let subscription_slot1 = 0;
+        let subscription_slot2 = 1;
+        let com1 = 1;
+        let com2 = 0;
+
+        // create the attestation service and subscriptions
+        let mut attestation_service = get_attestation_service();
+        let current_slot = attestation_service
+            .beacon_chain
+            .slot_clock
+            .now()
+            .expect("Could not get current slot");
+
+        let sub1 = get_subscription(
+            validator_index,
+            com1,
+            current_slot + Slot::new(subscription_slot1),
+            committee_count,
+        );
+
+        let sub2 = get_subscription(
+            validator_index,
+            com2,
+            current_slot + Slot::new(subscription_slot2),
+            committee_count,
+        );
+
+        let subnet_id1 = SubnetId::compute_subnet::<MainnetEthSpec>(
+            current_slot + Slot::new(subscription_slot1),
+            com1,
+            committee_count,
+            &attestation_service.beacon_chain.spec,
+        )
+        .unwrap();
+
+        let subnet_id2 = SubnetId::compute_subnet::<MainnetEthSpec>(
+            current_slot + Slot::new(subscription_slot2),
+            com2,
+            committee_count,
+            &attestation_service.beacon_chain.spec,
+        )
+        .unwrap();
+
+        // Assert that subscriptions are different but their subnet is the same
+        assert_ne!(sub1, sub2);
+        assert_eq!(subnet_id1, subnet_id2);
+
+        // submit the subscriptions
+        attestation_service
+            .validator_subscriptions(vec![sub1, sub2])
+            .unwrap();
+
+        // Unsubscription event should happen at slot 2 (since subnet id's are the same, unsubscription event should be at higher slot + 1)
+        // Get all events for 1 slot duration (unsubscription event should happen after 2 slot durations).
+        let events = get_events(&mut attestation_service, None, 1).await;
+        matches::assert_matches!(
+            events[..3],
+            [
+                SubnetServiceMessage::DiscoverPeers(_),
+                SubnetServiceMessage::Subscribe(_any1),
+                SubnetServiceMessage::EnrAdd(_any3)
+            ]
+        );
+
+        let expected = SubnetServiceMessage::Subscribe(Subnet::Attestation(subnet_id1));
+
+        // Should be still subscribed to 1 long lived and 1 short lived subnet if both are different.
+        if !attestation_service.random_subnets.contains(&subnet_id1) {
+            assert_eq!(expected, events[3]);
+            assert_eq!(attestation_service.subscription_count(), 2);
+        } else {
+            assert_eq!(attestation_service.subscription_count(), 1);
+        }
+
+        // Get event for 1 more slot duration, we should get the unsubscribe event now.
+        let unsubscribe_event = get_events(&mut attestation_service, None, 1).await;
+
+        // If the long lived and short lived subnets are different, we should get an unsubscription event.
+        if !attestation_service.random_subnets.contains(&subnet_id1) {
+            assert_eq!(
+                [SubnetServiceMessage::Unsubscribe(Subnet::Attestation(
+                    subnet_id1
+                ))],
+                unsubscribe_event[..]
+            );
+        }
+
+        // Should be subscribed to only 1 long lived subnet after unsubscription.
+        assert_eq!(attestation_service.subscription_count(), 1);
+    }
+
+    #[tokio::test]
+    async fn subscribe_all_random_subnets() {
+        let attestation_subnet_count = MainnetEthSpec::default_spec().attestation_subnet_count;
+        let subscription_slot = 10;
+        let subscription_count = attestation_subnet_count;
+        let committee_count = 1;
+
+        // create the attestation service and subscriptions
+        let mut attestation_service = get_attestation_service();
+        let current_slot = attestation_service
+            .beacon_chain
+            .slot_clock
+            .now()
+            .expect("Could not get current slot");
+
+        let subscriptions = get_subscriptions(
+            subscription_count,
+            current_slot + subscription_slot,
+            committee_count,
+        );
+
+        // submit the subscriptions
+        attestation_service
+            .validator_subscriptions(subscriptions)
+            .unwrap();
+
+        let events = get_events(&mut attestation_service, None, 3).await;
+        let mut discover_peer_count = 0;
+        let mut enr_add_count = 0;
+        let mut unexpected_msg_count = 0;
+
+        for event in &events {
+            match event {
+                SubnetServiceMessage::DiscoverPeers(_) => discover_peer_count += 1,
+                SubnetServiceMessage::Subscribe(_any_subnet) => {}
+                SubnetServiceMessage::EnrAdd(_any_subnet) => enr_add_count += 1,
+                _ => unexpected_msg_count += 1,
+            }
+        }
+
+        // The bulk discovery request length should be equal to validator_count
+        let bulk_discovery_event = events.last().unwrap();
+        if let SubnetServiceMessage::DiscoverPeers(d) = bulk_discovery_event {
+            assert_eq!(d.len(), attestation_subnet_count as usize);
+        } else {
+            panic!("Unexpected event {:?}", bulk_discovery_event);
+        }
+
+        // 64 `DiscoverPeer` requests of length 1 corresponding to random subnets
+        // and 1 `DiscoverPeer` request corresponding to bulk subnet discovery.
+        assert_eq!(discover_peer_count, subscription_count + 1);
+        assert_eq!(attestation_service.subscription_count(), 64);
+        assert_eq!(enr_add_count, 64);
+        assert_eq!(unexpected_msg_count, 0);
+        // test completed successfully
+    }
+
+    #[tokio::test]
+    async fn subscribe_all_random_subnets_plus_one() {
+        let attestation_subnet_count = MainnetEthSpec::default_spec().attestation_subnet_count;
+        let subscription_slot = 10;
+        // the 65th subscription should result in no more messages than the previous scenario
+        let subscription_count = attestation_subnet_count + 1;
+        let committee_count = 1;
+
+        // create the attestation service and subscriptions
+        let mut attestation_service = get_attestation_service();
+        let current_slot = attestation_service
+            .beacon_chain
+            .slot_clock
+            .now()
+            .expect("Could not get current slot");
+
+        let subscriptions = get_subscriptions(
+            subscription_count,
+            current_slot + subscription_slot,
+            committee_count,
+        );
+
+        // submit the subscriptions
+        attestation_service
+            .validator_subscriptions(subscriptions)
+            .unwrap();
+
+        let events = get_events(&mut attestation_service, None, 3).await;
+        let mut discover_peer_count = 0;
+        let mut enr_add_count = 0;
+        let mut unexpected_msg_count = 0;
+
+        for event in &events {
+            match event {
+                SubnetServiceMessage::DiscoverPeers(_) => discover_peer_count += 1,
+                SubnetServiceMessage::Subscribe(_any_subnet) => {}
+                SubnetServiceMessage::EnrAdd(_any_subnet) => enr_add_count += 1,
+                _ => unexpected_msg_count += 1,
+            }
+        }
+
+        // The bulk discovery request length shouldn't exceed max attestation_subnet_count
+        let bulk_discovery_event = events.last().unwrap();
+        if let SubnetServiceMessage::DiscoverPeers(d) = bulk_discovery_event {
+            assert_eq!(d.len(), attestation_subnet_count as usize);
+        } else {
+            panic!("Unexpected event {:?}", bulk_discovery_event);
+        }
+        // 64 `DiscoverPeer` requests of length 1 corresponding to random subnets
+        // and 1 `DiscoverPeer` request corresponding to the bulk subnet discovery.
+        // For the 65th subscription, the call to `subscribe_to_random_subnets` is not made because we are at capacity.
+        assert_eq!(discover_peer_count, 64 + 1);
+        assert_eq!(attestation_service.subscription_count(), 64);
+        assert_eq!(enr_add_count, 64);
+        assert_eq!(unexpected_msg_count, 0);
+    }
+}
+
+mod sync_committee_service {
+    use super::*;
+
+    #[tokio::test]
+    async fn subscribe_and_unsubscribe() {
+        // subscription config
+        let validator_index = 1;
+        let until_epoch = Epoch::new(1);
+        let sync_committee_indices = vec![1];
+
+        // create the attestation service and subscriptions
+        let mut sync_committee_service = get_sync_committee_service();
+
+        let subscriptions = vec![SyncCommitteeSubscription {
+            validator_index,
+            sync_committee_indices: sync_committee_indices.clone(),
+            until_epoch,
+        }];
+
+        // submit the subscriptions
+        sync_committee_service
+            .validator_subscriptions(subscriptions)
+            .unwrap();
+
+        let subnet_ids = SyncSubnetId::compute_subnets_for_sync_committee::<MainnetEthSpec>(
+            &sync_committee_indices,
+        )
+        .unwrap();
+        let subnet_id = subnet_ids.iter().next().unwrap();
+
+        // Note: the unsubscription event takes 2 epochs (8 * 2 * 0.4 secs = 3.2 secs)
+        let events = get_events(
+            &mut sync_committee_service,
+            Some(5),
+            (MainnetEthSpec::slots_per_epoch() * 3) as u32, // Have some buffer time before getting 5 events
+        )
+        .await;
+        assert_eq!(
+            events[..2],
+            [
+                SubnetServiceMessage::Subscribe(Subnet::SyncCommittee(*subnet_id)),
+                SubnetServiceMessage::EnrAdd(Subnet::SyncCommittee(*subnet_id))
+            ]
+        );
+        matches::assert_matches!(
+            events[2..],
+            [
+                SubnetServiceMessage::DiscoverPeers(_),
+                SubnetServiceMessage::Unsubscribe(_),
+                SubnetServiceMessage::EnrRemove(_),
+            ]
+        );
+
+        // Should be unsubscribed at the end.
+        assert_eq!(sync_committee_service.subscription_count(), 0);
+    }
+
+    #[tokio::test]
+    async fn same_subscription_with_lower_until_epoch() {
+        // subscription config
+        let validator_index = 1;
+        let until_epoch = Epoch::new(2);
+        let sync_committee_indices = vec![1];
+
+        // create the attestation service and subscriptions
+        let mut sync_committee_service = get_sync_committee_service();
+
+        let subscriptions = vec![SyncCommitteeSubscription {
+            validator_index,
+            sync_committee_indices: sync_committee_indices.clone(),
+            until_epoch,
+        }];
+
+        // submit the subscriptions
+        sync_committee_service
+            .validator_subscriptions(subscriptions)
+            .unwrap();
+
+        // Get all immediate events (won't include unsubscriptions)
+        let events = get_events(&mut sync_committee_service, None, 1).await;
+        matches::assert_matches!(
+            events[..],
+            [
+                SubnetServiceMessage::Subscribe(Subnet::SyncCommittee(_)),
+                SubnetServiceMessage::EnrAdd(Subnet::SyncCommittee(_)),
+                SubnetServiceMessage::DiscoverPeers(_),
+            ]
+        );
+
+        // Additional subscriptions which shouldn't emit any non-discovery events
+        // Event 1 is a duplicate of an existing subscription
+        // Event 2 is the same subscription with lower `until_epoch` than the existing subscription
+        let subscriptions = vec![
+            SyncCommitteeSubscription {
+                validator_index,
+                sync_committee_indices: sync_committee_indices.clone(),
+                until_epoch,
+            },
+            SyncCommitteeSubscription {
+                validator_index,
+                sync_committee_indices: sync_committee_indices.clone(),
+                until_epoch: until_epoch - 1,
+            },
+        ];
+
+        // submit the subscriptions
+        sync_committee_service
+            .validator_subscriptions(subscriptions)
+            .unwrap();
+
+        // Get all immediate events (won't include unsubscriptions)
+        let events = get_events(&mut sync_committee_service, None, 1).await;
+        matches::assert_matches!(events[..], [SubnetServiceMessage::DiscoverPeers(_),]);
+
+        // Should be unsubscribed at the end.
+        assert_eq!(sync_committee_service.subscription_count(), 1);
+    }
+}