only initially subscribe to relevant attestation subnets (#2231)

beacon_chain/nimbus_beacon_node.nim

@@ -1,5 +1,5 @@
 # beacon_chain
-# Copyright (c) 2018-2020 Status Research & Development GmbH
+# Copyright (c) 2018-2021 Status Research & Development GmbH
 # Licensed and distributed under either of
 #   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
 #   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
@@ -375,6 +375,12 @@ func getStabilitySubnets(stabilitySubnets: auto): set[uint8] =
   for subnetInfo in stabilitySubnets:
     result.incl subnetInfo.subnet
 
+proc getAttachedValidators(node: BeaconNode): seq[ValidatorIndex] =
+  for validatorIndex in 0 ..< node.chainDag.headState.data.data.validators.len:
+    if node.getAttachedValidator(
+        node.chainDag.headState.data.data, validatorIndex.ValidatorIndex) != nil:
+      result.add validatorIndex.ValidatorIndex
+
 proc cycleAttestationSubnets(node: BeaconNode, wallSlot: Slot) =
   static: doAssert RANDOM_SUBNETS_PER_VALIDATOR == 1
   doAssert not node.config.subscribeAllSubnets
@@ -401,12 +407,9 @@ proc cycleAttestationSubnets(node: BeaconNode, wallSlot: Slot) =
   # important to attest regardless, as those upcoming blocks will hit maximum
   # attestations quickly and any individual attestation's likelihood of being
   # selected is low.
-  let epochParity = wallSlot.epoch mod 2
+  let
-  var attachedValidators: seq[ValidatorIndex]
+    epochParity = wallSlot.epoch mod 2
-  for validatorIndex in 0 ..< node.chainDag.headState.data.data.validators.len:
+    attachedValidators = node.getAttachedValidators()
-    if node.getAttachedValidator(
-        node.chainDag.headState.data.data, validatorIndex.ValidatorIndex) != nil:
-      attachedValidators.add validatorIndex.ValidatorIndex
 
   let (newAttestationSubnets, expiringSubnets, newSubnets) =
     get_attestation_subnet_changes(
@@ -441,26 +444,55 @@ proc cycleAttestationSubnets(node: BeaconNode, wallSlot: Slot) =
     if stabilitySubnets != prevStabilitySubnets:
       node.updateStabilitySubnetMetadata(stabilitySubnets)
 
-proc getAttestationSubnetHandlers(node: BeaconNode) =
+proc getInitialAttestationSubnets(node: BeaconNode): set[uint8] =
-  var initialSubnets: set[uint8]
+  let
-  # If/when this stops subscribing to all attestation subnet handlers, the
+    wallEpoch = node.beaconClock.now().slotOrZero().epoch
-  # subscribeAllSubnets implementation needs modification from its current
+    validator_indices = toHashSet(node.getAttachedValidators())
-  # no-op/exploit-defaults version.
-  for i in 0'u8 ..< ATTESTATION_SUBNET_COUNT:
-    initialSubnets.incl i
 
+  template mergeAttestationSubnets(epoch: Epoch) =
+    for (subnetIndex, _) in get_committee_assignments(
+        node.chainDag.headState.data.data, epoch, validator_indices):
+      result.incl subnetIndex
+
+  # Either wallEpoch is 0, in which case it might be pre-genesis, but we only
+  # care about the already-known first two epochs of attestations, or it's in
+  # epoch 0 for real, in which case both are also already known; or wallEpoch
+  # is greater than 0, in which case it's being called from onSlotStart which
+  # has enough state to calculate wallEpoch + {0,1}'s attestations.
+  mergeAttestationSubnets(wallEpoch)
+  mergeAttestationSubnets(wallEpoch + 1)
+
+proc getAttestationSubnetHandlers(node: BeaconNode) =
   # https://github.com/ethereum/eth2.0-specs/blob/v1.0.0/specs/phase0/validator.md#phase-0-attestation-subnet-stability
   # TODO:
   # We might want to reuse the previous stability subnet if not expired when:
   # - Restarting the node with a presistent netkey
   # - When going from synced -> syncing -> synced state
-  let wallEpoch = node.beaconClock.now().slotOrZero().epoch
+
+  template getAllAttestationSubnets(): set[uint8] =
+    var subnets: set[uint8]
+    for i in 0'u8 ..< ATTESTATION_SUBNET_COUNT:
+      subnets.incl i
+    subnets
+
+  let
+    initialSubnets =
+      if node.config.subscribeAllSubnets:
+        getAllAttestationSubnets()
+      else:
+        node.getInitialAttestationSubnets()
+    wallEpoch = node.beaconClock.now().slotOrZero().epoch
+
+  var initialStabilitySubnets: set[uint8]
+
   node.attestationSubnets.stabilitySubnets.setLen(
     node.attachedValidators.count)
   for i in 0 ..< node.attachedValidators.count:
     node.attestationSubnets.stabilitySubnets[i] = (
       subnet: rand(ATTESTATION_SUBNET_COUNT - 1).uint8,
       expiration: wallEpoch + getStabilitySubnetLength())
+    initialStabilitySubnets.incl(
+      node.attestationSubnets.stabilitySubnets[i].subnet)
 
   node.updateStabilitySubnetMetadata(
     if node.config.subscribeAllSubnets:
@@ -478,8 +510,10 @@ proc getAttestationSubnetHandlers(node: BeaconNode) =
 
   debug "Initial attestation subnets subscribed",
      initialSubnets,
+     initialStabilitySubnets,
      wallEpoch
-  node.installAttestationSubnetHandlers(initialSubnets)
+  node.installAttestationSubnetHandlers(
+    initialSubnets + initialStabilitySubnets)
 
 proc addMessageHandlers(node: BeaconNode) =
   node.network.subscribe(node.topicBeaconBlocks, enableTopicMetrics = true)
@@ -857,8 +891,9 @@ proc run*(node: BeaconNode) =
     node.requestManager.start()
     node.startSyncManager()
 
-    node.addMessageHandlers()
+    if not node.beaconClock.now().toSlot().afterGenesis:
-    doAssert node.getTopicSubscriptionEnabled()
+      node.addMessageHandlers()
+      doAssert node.getTopicSubscriptionEnabled()
 
   ## Ctrl+C handling
   proc controlCHandler() {.noconv.} =

beacon_chain/spec/network.nim

@@ -1,5 +1,5 @@
 # beacon_chain
-# Copyright (c) 2018-2020 Status Research & Development GmbH
+# Copyright (c) 2018-2021 Status Research & Development GmbH
 # Licensed and distributed under either of
 #   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
 #   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
@@ -89,10 +89,10 @@ func getAttestationTopic*(forkDigest: ForkDigest, subnetIndex: uint64):
   except ValueError as e:
     raiseAssert e.msg
 
-func get_committee_assignments(
+func get_committee_assignments*(
     state: BeaconState, epoch: Epoch,
     validator_indices: HashSet[ValidatorIndex]):
-    seq[tuple[subnetIndex: uint64, slot: Slot]] =
+    seq[tuple[subnetIndex: uint8, slot: Slot]] =
   var cache = StateCache()
 
   let
@@ -105,7 +105,7 @@ func get_committee_assignments(
       if not disjoint(validator_indices,
           get_beacon_committee(state, slot, idx, cache).toHashSet):
         result.add(
-          (compute_subnet_for_attestation(committees_per_slot, slot, idx),
+          (compute_subnet_for_attestation(committees_per_slot, slot, idx).uint8,
             slot))
 
 # https://github.com/ethereum/eth2.0-specs/blob/v1.0.0/specs/phase0/validator.md#phase-0-attestation-subnet-stability