fix subnet calculation in RPC and insert broadcast attestations into node's pool (#2207)

* fix subnet calculation in RPC and insert broadcast attestations into node's pool

* unify codepaths to ensure only mostly-checked-to-be-valid attestations enter the pool, even from node's own broadcasts

* update attestation pool tests for new validateAttestation param

beacon_chain/attestation_aggregation.nim

@@ -162,7 +162,7 @@ func check_attestation_subnet(
 proc validateAttestation*(
     pool: var AttestationPool,
     attestation: Attestation, wallTime: BeaconTime,
-    topicCommitteeIndex: uint64):
+    topicCommitteeIndex: uint64, checksExpensive: bool):
     Result[HashSet[ValidatorIndex], (ValidationResult, cstring)] =
   # [REJECT] The attestation's epoch matches its target -- i.e.
   # attestation.data.target.epoch ==
@@ -258,6 +258,9 @@ proc validateAttestation*(
     return err((ValidationResult.Ignore, cstring(
       "Validator has already voted in epoch")))
 
+  if not checksExpensive:
+    return ok(attesting_indices)
+
   # The signature of attestation is valid.
   block:
     let v = is_valid_indexed_attestation(

beacon_chain/eth2_processor.nim

@@ -288,7 +288,8 @@ proc blockValidator*(
 proc attestationValidator*(
     self: var Eth2Processor,
     attestation: Attestation,
-    committeeIndex: uint64): ValidationResult =
+    committeeIndex: uint64,
+    checksExpensive: bool = true): ValidationResult =
   logScope:
     attestation = shortLog(attestation)
     committeeIndex
@@ -307,7 +308,7 @@ proc attestationValidator*(
   let delay = wallTime - attestation.data.slot.toBeaconTime
   debug "Attestation received", delay
   let v = self.attestationPool[].validateAttestation(
-      attestation, wallTime, committeeIndex)
+      attestation, wallTime, committeeIndex, checksExpensive)
   if v.isErr():
     debug "Dropping attestation", err = v.error()
     return v.error[0]

beacon_chain/nimbus_validator_client.nim

@@ -181,8 +181,6 @@ proc onSlotStart(vc: ValidatorClient, lastSlot, scheduledSlot: Slot) {.gcsafe, a
         vc.attestationsForEpoch[epoch].contains slot:
       var validatorToAttestationDataRoot: Table[ValidatorPubKey, Eth2Digest]
       for a in vc.attestationsForEpoch[epoch][slot]:
-        notice "Attesting", slot = slot, public_key = a.public_key
-
         let validator = vc.attachedValidators.validators[a.public_key]
         let ad = await vc.client.get_v1_validator_attestation_data(slot, a.committee_index)
 
@@ -206,6 +204,8 @@ proc onSlotStart(vc: ValidatorClient, lastSlot, scheduledSlot: Slot) {.gcsafe, a
             ad, a.committee_length.int, a.validator_committee_index,
             vc.fork, vc.beaconGenesis.genesis_validators_root)
 
+          notice "Attesting",
+            slot, public_key = a.public_key, attestation = shortLog(attestation)
           discard await vc.client.post_v1_beacon_pool_attestations(attestation)
 
           validatorToAttestationDataRoot[a.public_key] = attestation.data.hash_tree_root

beacon_chain/rpc/validator_api.nim

@@ -142,7 +142,11 @@ proc installValidatorApiHandlers*(rpcServer: RpcServer, node: BeaconNode) =
       raise newException(CatchableError,
         "Invalid slot signature")
 
-    let subnet = committee_index.uint8
+    let
+      head = node.doChecksAndGetCurrentHead(epoch)
+      epochRef = node.chainDag.getEpochRef(head, epoch)
+    let subnet = compute_subnet_for_attestation(
+      get_committee_count_per_slot(epochRef), slot, committee_index).uint8
     if  subnet notin node.attestationSubnets.subscribedSubnets[0] and
         subnet notin node.attestationSubnets.subscribedSubnets[1]:
       await node.network.subscribe(getAttestationTopic(

beacon_chain/spec/network.nim

@@ -89,15 +89,6 @@ func getAttestationTopic*(forkDigest: ForkDigest, subnetIndex: uint64):
   except ValueError as e:
     raiseAssert e.msg
 
-func getAttestationTopic*(forkDigest: ForkDigest,
-                          attestation: Attestation,
-                          num_active_validators: uint64): string =
-  getAttestationTopic(
-    forkDigest,
-    compute_subnet_for_attestation(
-      get_committee_count_per_slot(num_active_validators),
-      attestation.data.slot, attestation.data.index.CommitteeIndex))
-
 func get_committee_assignments(
     state: BeaconState, epoch: Epoch,
     validator_indices: HashSet[ValidatorIndex]):

beacon_chain/validator_duties.nim

@@ -19,7 +19,8 @@ import
 
   # Local modules
   ./spec/[
-    datatypes, digest, crypto, helpers, network, signatures, state_transition],
+    datatypes, digest, crypto, helpers, network, signatures, state_transition,
+    validator],
   ./conf, ./time, ./validator_pool,
   ./attestation_pool, ./exit_pool,
   ./block_pools/[spec_cache, chain_dag, clearance],
@@ -152,9 +153,16 @@ proc isSynced*(node: BeaconNode, head: BlockRef): bool =
 
 proc sendAttestation*(
     node: BeaconNode, attestation: Attestation, num_active_validators: uint64) =
+  let subnet_index =
+    compute_subnet_for_attestation(
+      get_committee_count_per_slot(num_active_validators), attestation.data.slot,
+      attestation.data.index.CommitteeIndex)
   node.network.broadcast(
-    getAttestationTopic(node.forkDigest, attestation, num_active_validators),
-    attestation)
+    getAttestationTopic(node.forkDigest, subnet_index), attestation)
+
+  # Ensure node's own broadcast attestations end up in its attestation pool
+  discard node.processor[].attestationValidator(
+    attestation, subnet_index, false)
 
   beacon_attestations_sent.inc()
 

tests/test_attestation_pool.nim

@@ -438,22 +438,22 @@ suiteReport "Attestation validation " & preset():
       beaconTime = attestation.data.slot.toBeaconTime()
 
     check:
-      validateAttestation(pool[], attestation, beaconTime, subnet).isOk
+      validateAttestation(pool[], attestation, beaconTime, subnet, true).isOk
 
       # Same validator again
-      validateAttestation(pool[], attestation, beaconTime, subnet).error()[0] ==
+      validateAttestation(pool[], attestation, beaconTime, subnet, true).error()[0] ==
         ValidationResult.Ignore
 
     pool[].nextAttestationEpoch.setLen(0) # reset for test
     check:
       # Wrong subnet
-      validateAttestation(pool[], attestation, beaconTime, subnet + 1).isErr
+      validateAttestation(pool[], attestation, beaconTime, subnet + 1, true).isErr
 
     pool[].nextAttestationEpoch.setLen(0) # reset for test
     check:
       # Too far in the future
       validateAttestation(
-        pool[], attestation, beaconTime - 1.seconds, subnet + 1).isErr
+        pool[], attestation, beaconTime - 1.seconds, subnet + 1, true).isErr
 
     pool[].nextAttestationEpoch.setLen(0) # reset for test
     check:
@@ -461,4 +461,4 @@ suiteReport "Attestation validation " & preset():
       validateAttestation(
         pool[], attestation,
         beaconTime - (SECONDS_PER_SLOT * SLOTS_PER_EPOCH - 1).int.seconds,
-        subnet + 1).isErr
+        subnet + 1, true).isErr