From 92e7e288e749da6cf23a4596efd76e309af42ee2 Mon Sep 17 00:00:00 2001
From: Jacek Sieka <jacek@status.im>
Date: Fri, 25 Feb 2022 17:15:39 +0100
Subject: [PATCH] Ignore seen aggregates (#3439)

https://github.com/ethereum/consensus-specs/pull/2225 removed an ignore
rule that would filter out duplicate aggregates from gossip publishing -
however, this causes increased bandwidth and CPU usage as discussed in
https://github.com/ethereum/consensus-specs/issues/2183 - the intent is
to revert the removal and reinstate the rule.

This PR implements ignore filtering which cuts down on CPU usage (fewer
aggregates to validate) and bandwidth usage (less fanout of duplicates)
- as #2225 points out, this may lead to a small increase in IHAVE
messages.
---
 .../attestation_pool.nim                      | 32 +++++++++++++++----
 .../gossip_processing/gossip_validation.nim   | 24 ++++++++++++--
 tests/test_attestation_pool.nim               |  4 +++
 3 files changed, 51 insertions(+), 9 deletions(-)

diff --git a/beacon_chain/consensus_object_pools/attestation_pool.nim b/beacon_chain/consensus_object_pools/attestation_pool.nim
index 6a995da4d..711ae9279 100644
--- a/beacon_chain/consensus_object_pools/attestation_pool.nim
+++ b/beacon_chain/consensus_object_pools/attestation_pool.nim
@@ -280,6 +280,12 @@ func updateAggregates(entry: var AttestationEntry) =
             inc j
         inc i
 
+func covers(entry: AttestationEntry, bits: CommitteeValidatorsBits): bool =
+  for i in 0..<entry.aggregates.len():
+    if bits.isSubsetOf(entry.aggregates[i].aggregation_bits):
+      return true
+  false
+
 proc addAttestation(entry: var AttestationEntry,
                     attestation: Attestation,
                     signature: CookedSig): bool =
@@ -304,12 +310,8 @@ proc addAttestation(entry: var AttestationEntry,
     entry.singles[singleIndex.get()] = signature
   else:
     # More than one vote in this attestation
-    for i in 0..<entry.aggregates.len():
-      if attestation.aggregation_bits.isSubsetOf(entry.aggregates[i].aggregation_bits):
-        trace "Aggregate already seen",
-          singles = entry.singles.len(),
-          aggregates = entry.aggregates.len()
-        return false
+    if entry.covers(attestation.aggregation_bits):
+      return false
 
     # Since we're adding a new aggregate, we can now remove existing
     # aggregates that don't add any new votes
@@ -376,6 +378,24 @@ proc addAttestation*(pool: var AttestationPool,
   if not(isNil(pool.onAttestationAdded)):
     pool.onAttestationAdded(attestation)
 
+func covers*(
+    pool: var AttestationPool, data: Attestationdata,
+    bits: CommitteeValidatorsBits): bool =
+  ## Return true iff the given attestation already is fully covered by one of
+  ## the existing aggregates, making it redundant
+  ## the `var` attestation pool is needed to use `withValue`, else Table becomes
+  ## unusably inefficient
+  let candidateIdx = pool.candidateIdx(data.slot)
+  if candidateIdx.isNone:
+    return false
+
+  let attestation_data_root = hash_tree_root(data)
+  pool.candidates[candidateIdx.get()].withValue(attestation_data_root, entry):
+    if entry[].covers(bits):
+      return true
+
+  false
+
 proc addForkChoice*(pool: var AttestationPool,
                     epochRef: EpochRef,
                     blckRef: BlockRef,
diff --git a/beacon_chain/gossip_processing/gossip_validation.nim b/beacon_chain/gossip_processing/gossip_validation.nim
index 4f2e63ec4..d8bd298d2 100644
--- a/beacon_chain/gossip_processing/gossip_validation.nim
+++ b/beacon_chain/gossip_processing/gossip_validation.nim
@@ -621,9 +621,24 @@ proc validateAggregate*(
       return err(v.error)
     v.get()
 
-  # [REJECT] aggregate_and_proof.selection_proof selects the validator as an
-  # aggregator for the slot -- i.e. is_aggregator(state, aggregate.data.slot,
-  # aggregate.data.index, aggregate_and_proof.selection_proof) returns True.
+  if pool[].covers(aggregate.data, aggregate.aggregation_bits):
+    # https://github.com/ethereum/consensus-specs/issues/2183 - althoughh this
+    # check was temporarily removed from the spec, the intent is to reinstate it
+    # per discussion in the ticket.
+    #
+    # [IGNORE] The valid aggregate attestation defined by
+    # `hash_tree_root(aggregate)` has _not_ already been seen
+    # (via aggregate gossip, within a verified block, or through the creation of
+    # an equivalent aggregate locally).
+
+    # Our implementation of this check is slightly different in that it doesn't
+    # consider aggregates from verified blocks - this would take a rather heavy
+    # index to work correcly under fork conditions - we also check for coverage
+    # of attestation bits instead of comparing with full root of aggreagte:
+    # this captures the spirit of the checkk by ignoring aggregates that are
+    # strict subsets of other, already-seen aggregates.
+    return errIgnore("Aggregate already covered")
+
   let
     epochRef = block:
       let tmp = pool.dag.getEpochRef(target.blck, target.slot.epoch, false)
@@ -641,6 +656,9 @@ proc validateAggregate*(
       return checkedReject("Attestation: committee index not within expected range")
     idx.get()
 
+  # [REJECT] aggregate_and_proof.selection_proof selects the validator as an
+  # aggregator for the slot -- i.e. is_aggregator(state, aggregate.data.slot,
+  # aggregate.data.index, aggregate_and_proof.selection_proof) returns True.
   if not is_aggregator(
       epochRef, slot, committee_index, aggregate_and_proof.selection_proof):
     return checkedReject("Aggregate: incorrect aggregator")
diff --git a/tests/test_attestation_pool.nim b/tests/test_attestation_pool.nim
index e77890581..ce41aa2a8 100644
--- a/tests/test_attestation_pool.nim
+++ b/tests/test_attestation_pool.nim
@@ -202,6 +202,9 @@ suite "Attestation pool processing" & preset():
     att0.combine(att2)
     att1.combine(att2)
 
+    check:
+      not pool[].covers(att0.data, att0.aggregation_bits)
+
     pool[].addAttestation(
       att0, @[bc0[0], bc0[2]], att0.loadSig, att0.data.slot.start_beacon_time)
     pool[].addAttestation(
@@ -214,6 +217,7 @@ suite "Attestation pool processing" & preset():
         info, {}).isOk()
 
     check:
+      pool[].covers(att0.data, att0.aggregation_bits)
       pool[].getAttestationsForBlock(state.data, cache).len() == 2
       # Can get either aggregate here, random!
       pool[].getAggregatedAttestation(1.Slot, 0.CommitteeIndex).isSome()