Ignore gossip AggregateAndProof if aggregate is seen

packages/fork-choice/src/forkChoice/forkChoice.ts

@@ -444,7 +444,7 @@ export class ForkChoice implements IForkChoice {
    * The supplied `attestation` **must** pass the `in_valid_indexed_attestation` function as it
    * will not be run here.
    */
-  onAttestation(attestation: phase0.IndexedAttestation): void {
+  onAttestation(attestation: phase0.IndexedAttestation, attDataRoot?: string): void {
     // Ignore any attestations to the zero hash.
     //
     // This is an edge case that results from the spec aliasing the zero hash to the genesis
@@ -466,7 +466,7 @@ export class ForkChoice implements IForkChoice {
       return;
     }
 
-    this.validateOnAttestation(attestation, slot, blockRootHex, targetEpoch);
+    this.validateOnAttestation(attestation, slot, blockRootHex, targetEpoch, attDataRoot);
 
     if (slot < this.fcStore.currentSlot) {
       for (const validatorIndex of attestation.attestingIndices) {
@@ -789,7 +789,8 @@ export class ForkChoice implements IForkChoice {
     indexedAttestation: phase0.IndexedAttestation,
     slot: Slot,
     blockRootHex: string,
-    targetEpoch: Epoch
+    targetEpoch: Epoch,
+    attDataRoot?: string
   ): void {
     // There is no point in processing an attestation with an empty bitfield. Reject
     // it immediately.
@@ -807,7 +808,7 @@ export class ForkChoice implements IForkChoice {
 
     const attestationData = indexedAttestation.data;
     // AttestationData is expected to internally cache its root to make this hashTreeRoot() call free
-    const attestationCacheKey = toHexString(ssz.phase0.AttestationData.hashTreeRoot(attestationData));
+    const attestationCacheKey = attDataRoot ?? toHexString(ssz.phase0.AttestationData.hashTreeRoot(attestationData));
 
     if (!this.validatedAttestationDatas.has(attestationCacheKey)) {
       this.validateAttestationData(indexedAttestation.data, slot, blockRootHex, targetEpoch, attestationCacheKey);

packages/fork-choice/src/forkChoice/interface.ts

@@ -77,7 +77,7 @@ export interface IForkChoice {
    * The supplied `attestation` **must** pass the `in_valid_indexed_attestation` function as it
    * will not be run here.
    */
-  onAttestation(attestation: phase0.IndexedAttestation): void;
+  onAttestation(attestation: phase0.IndexedAttestation, attDataRoot?: string): void;
   getLatestMessage(validatorIndex: ValidatorIndex): ILatestMessage | undefined;
   /**
    * Call `onTick` for all slots between `fcStore.getCurrentSlot()` and the provided `currentSlot`.

packages/lodestar/src/chain/blocks/importBlock.ts

@@ -1,3 +1,4 @@
+import {ssz} from "@chainsafe/lodestar-types";
 import {SLOTS_PER_EPOCH} from "@chainsafe/lodestar-params";
 import {toHexString} from "@chainsafe/ssz";
 import {allForks} from "@chainsafe/lodestar-types";
@@ -23,7 +24,7 @@ import {LightClientServer} from "../lightClient";
 import {getCheckpointFromState} from "./utils/checkpoint";
 import {PendingEvents} from "./utils/pendingEvents";
 import {FullyVerifiedBlock} from "./types";
-// import {ForkChoiceError, ForkChoiceErrorCode} from "@chainsafe/lodestar-fork-choice/lib/forkChoice/errors";
+import {SeenAggregatedAttestations} from "../seenCache/seenAggregateAndProof";
 
 /**
  * Fork-choice allows to import attestations from current (0) or past (1) epoch.
@@ -35,6 +36,7 @@ export type ImportBlockModules = {
   forkChoice: IForkChoice;
   stateCache: StateContextCache;
   checkpointStateCache: CheckpointStateCache;
+  seenAggregatedAttestations: SeenAggregatedAttestations;
   lightClientServer: LightClientServer;
   executionEngine: IExecutionEngine;
   emitter: ChainEventEmitter;
@@ -120,10 +122,17 @@ export async function importBlock(chain: ImportBlockModules, fullyVerifiedBlock:
         const indexedAttestation = postState.epochCtx.getIndexedAttestation(attestation);
         const targetEpoch = attestation.data.target.epoch;
 
+        const attDataRoot = toHexString(ssz.phase0.AttestationData.hashTreeRoot(indexedAttestation.data));
+        chain.seenAggregatedAttestations.add(
+          targetEpoch,
+          attDataRoot,
+          {aggregationBits: attestation.aggregationBits, trueBitCount: indexedAttestation.attestingIndices.length},
+          true
+        );
         // Duplicated logic from fork-choice onAttestation validation logic.
         // Attestations outside of this range will be dropped as Errors, so no need to import
         if (targetEpoch <= currentEpoch && targetEpoch >= currentEpoch - FORK_CHOICE_ATT_EPOCH_LIMIT) {
-          chain.forkChoice.onAttestation(indexedAttestation);
+          chain.forkChoice.onAttestation(indexedAttestation, attDataRoot);
         }
 
         if (parentSlot !== undefined) {

packages/lodestar/src/chain/chain.ts

@@ -51,6 +51,7 @@ import {IEth1ForBlockProduction} from "../eth1";
 import {IExecutionEngine} from "../executionEngine";
 import {PrecomputeNextEpochTransitionScheduler} from "./precomputeNextEpochTransition";
 import {ReprocessController} from "./reprocess";
+import {SeenAggregatedAttestations} from "./seenCache/seenAggregateAndProof";
 
 export class BeaconChain implements IBeaconChain {
   readonly genesisTime: UintNum64;
@@ -81,9 +82,10 @@ export class BeaconChain implements IBeaconChain {
   // Gossip seen cache
   readonly seenAttesters = new SeenAttesters();
   readonly seenAggregators = new SeenAggregators();
+  readonly seenAggregatedAttestations: SeenAggregatedAttestations;
   readonly seenBlockProposers = new SeenBlockProposers();
   readonly seenSyncCommitteeMessages = new SeenSyncCommitteeMessages();
-  readonly seenContributionAndProof = new SeenContributionAndProof();
+  readonly seenContributionAndProof: SeenContributionAndProof;
 
   // Global state caches
   readonly pubkey2index: PubkeyIndexMap;
@@ -139,6 +141,9 @@ export class BeaconChain implements IBeaconChain {
     const stateCache = new StateContextCache({metrics});
     const checkpointStateCache = new CheckpointStateCache({metrics});
 
+    this.seenAggregatedAttestations = new SeenAggregatedAttestations(metrics);
+    this.seenContributionAndProof = new SeenContributionAndProof(metrics);
+
     // Initialize single global instance of state caches
     this.pubkey2index = new PubkeyIndexMap();
     this.index2pubkey = [];
@@ -188,6 +193,7 @@ export class BeaconChain implements IBeaconChain {
         lightClientServer,
         stateCache,
         checkpointStateCache,
+        seenAggregatedAttestations: this.seenAggregatedAttestations,
         emitter,
         config,
         logger,

packages/lodestar/src/chain/errors/attestationError.ts

@@ -37,6 +37,10 @@ export enum AttestationErrorCode {
    * There has already been an aggregation observed for this validator, we refuse to process a second.
    */
   AGGREGATOR_ALREADY_KNOWN = "ATTESTATION_ERROR_AGGREGATOR_ALREADY_KNOWN",
+  /**
+   * All of the attesters are known, we refuse to process subset of attesting indices since it brings no value.
+   */
+  ATTESTERS_ALREADY_KNOWN = "ATTESTATION_ERROR_ATTESTERS_ALREADY_KNOWN",
   /**
    * The aggregator index is higher than the maximum possible validator count.
    */
@@ -133,6 +137,7 @@ export type AttestationErrorType =
   | {code: AttestationErrorCode.AGGREGATOR_PUBKEY_UNKNOWN; aggregatorIndex: ValidatorIndex}
   | {code: AttestationErrorCode.ATTESTATION_ALREADY_KNOWN; targetEpoch: Epoch; validatorIndex: number}
   | {code: AttestationErrorCode.AGGREGATOR_ALREADY_KNOWN; targetEpoch: Epoch; aggregatorIndex: number}
+  | {code: AttestationErrorCode.ATTESTERS_ALREADY_KNOWN; targetEpoch: Epoch; aggregateRoot: RootHex}
   | {code: AttestationErrorCode.AGGREGATOR_INDEX_TOO_HIGH; aggregatorIndex: ValidatorIndex}
   | {code: AttestationErrorCode.UNKNOWN_OR_PREFINALIZED_BEACON_BLOCK_ROOT; root: RootHex}
   | {code: AttestationErrorCode.BAD_TARGET_EPOCH}

packages/lodestar/src/chain/errors/syncCommitteeError.ts

@@ -4,7 +4,8 @@ import {GossipActionError} from "./gossipValidation";
 export enum SyncCommitteeErrorCode {
   NOT_CURRENT_SLOT = "SYNC_COMMITTEE_ERROR_NOT_CURRENT_SLOT",
   UNKNOWN_BEACON_BLOCK_ROOT = "SYNC_COMMITTEE_ERROR_UNKNOWN_BEACON_BLOCK_ROOT",
-  SYNC_COMMITTEE_ALREADY_KNOWN = "SYNC_COMMITTEE_ERROR_SYNC_COMMITTEE_ALREADY_KNOWN",
+  SYNC_COMMITTEE_AGGREGATOR_ALREADY_KNOWN = "SYNC_COMMITTEE_ERROR_SYNC_COMMITTEE_AGGREGATOR_ALREADY_KNOWN",
+  SYNC_COMMITTEE_PARTICIPANTS_ALREADY_KNOWN = "SYNC_COMMITTEE_ERROR_SYNC_COMMITTEE_PARTICIPANTS_ALREADY_KNOWN",
   VALIDATOR_NOT_IN_SYNC_COMMITTEE = "SYNC_COMMITTEE_ERROR_VALIDATOR_NOT_IN_SYNC_COMMITTEE",
   INVALID_SIGNATURE = "SYNC_COMMITTEE_INVALID_SIGNATURE",
   INVALID_SUBCOMMITTEE_INDEX = "SYNC_COMMITTEE_INVALID_SUBCOMMITTEE_INDEX",
@@ -15,7 +16,8 @@ export enum SyncCommitteeErrorCode {
 export type SyncCommitteeErrorType =
   | {code: SyncCommitteeErrorCode.NOT_CURRENT_SLOT; slot: Slot; currentSlot: Slot}
   | {code: SyncCommitteeErrorCode.UNKNOWN_BEACON_BLOCK_ROOT; beaconBlockRoot: Uint8Array}
-  | {code: SyncCommitteeErrorCode.SYNC_COMMITTEE_ALREADY_KNOWN}
+  | {code: SyncCommitteeErrorCode.SYNC_COMMITTEE_AGGREGATOR_ALREADY_KNOWN}
+  | {code: SyncCommitteeErrorCode.SYNC_COMMITTEE_PARTICIPANTS_ALREADY_KNOWN}
   | {code: SyncCommitteeErrorCode.VALIDATOR_NOT_IN_SYNC_COMMITTEE; validatorIndex: ValidatorIndex}
   | {code: SyncCommitteeErrorCode.INVALID_SIGNATURE}
   | {code: SyncCommitteeErrorCode.INVALID_SUBCOMMITTEE_INDEX; subcommitteeIndex: number}

packages/lodestar/src/chain/eventHandlers.ts

@@ -96,6 +96,7 @@ export async function onClockSlot(this: BeaconChain, slot: Slot): Promise<void>
 export function onClockEpoch(this: BeaconChain, currentEpoch: Epoch): void {
   this.seenAttesters.prune(currentEpoch);
   this.seenAggregators.prune(currentEpoch);
+  this.seenAggregatedAttestations.prune(currentEpoch);
 }
 
 export function onForkVersion(this: BeaconChain, version: Version): void {

packages/lodestar/src/chain/interface.ts

@@ -22,6 +22,7 @@ import {LightClientServer} from "./lightClient";
 import {AggregatedAttestationPool} from "./opPools/aggregatedAttestationPool";
 import {PartiallyVerifiedBlockFlags} from "./blocks/types";
 import {ReprocessController} from "./reprocess";
+import {SeenAggregatedAttestations} from "./seenCache/seenAggregateAndProof";
 
 export type Eth2Context = {
   activeValidatorCount: number;
@@ -64,6 +65,7 @@ export interface IBeaconChain {
   // Gossip seen cache
   readonly seenAttesters: SeenAttesters;
   readonly seenAggregators: SeenAggregators;
+  readonly seenAggregatedAttestations: SeenAggregatedAttestations;
   readonly seenBlockProposers: SeenBlockProposers;
   readonly seenSyncCommitteeMessages: SeenSyncCommitteeMessages;
   readonly seenContributionAndProof: SeenContributionAndProof;

packages/lodestar/src/chain/seenCache/seenAggregateAndProof.ts

@@ -0,0 +1,91 @@
+import {Epoch, RootHex} from "@chainsafe/lodestar-types";
+import {BitArray} from "@chainsafe/ssz";
+import {IMetrics} from "../../metrics";
+import {isSuperSetOrEqual} from "../../util/bitArray";
+import {MapDef} from "../../util/map";
+
+/**
+ * With this gossip validation condition: [IGNORE] aggregate.data.slot is within the last ATTESTATION_PROPAGATION_SLOT_RANGE slots (with a MAXIMUM_GOSSIP_CLOCK_DISPARITY allowance)
+ * Since ATTESTATION_PROPAGATION_SLOT_RANGE is 32, we keep seen AggregateAndProof in the last 2 epochs.
+ */
+const MAX_EPOCHS_IN_CACHE = 2;
+
+export type AggregationInfo = {
+  aggregationBits: BitArray;
+  trueBitCount: number;
+};
+
+/**
+ * Although there are up to TARGET_AGGREGATORS_PER_COMMITTEE (16 for mainnet) AggregateAndProof messages per slot,
+ * they tend to have the same aggregate attestation, or one attestation is non-strict superset of another,
+ * the gossipsub messages-ids are different because they are really different SignedAggregateAndProof object.
+ * This is used to address the following spec in p2p-interface gossipsub:
+ * _[IGNORE]_ A valid aggregate attestation defined by `hash_tree_root(aggregate.data)` whose `aggregation_bits` is a
+ * non-strict superset has _not_ already been seen.
+ *
+ * We have AggregatedAttestationPool op pool, however aggregated attestations are not added to that place while this does.
+ */
+export class SeenAggregatedAttestations {
+  /**
+   * Array of AttestingIndices by same attestation data root by epoch.
+   * Note that there are at most TARGET_AGGREGATORS_PER_COMMITTEE (16) per attestation data.
+   * */
+  private readonly aggregateRootsByEpoch = new MapDef<Epoch, MapDef<RootHex, AggregationInfo[]>>(
+    () => new MapDef<RootHex, AggregationInfo[]>(() => [])
+  );
+  private lowestPermissibleEpoch: Epoch = 0;
+
+  constructor(private readonly metrics: IMetrics | null) {}
+
+  isKnown(targetEpoch: Epoch, attDataRoot: RootHex, aggregationBits: BitArray): boolean {
+    const seenAggregationInfoArr = this.aggregateRootsByEpoch.getOrDefault(targetEpoch).getOrDefault(attDataRoot);
+    this.metrics?.seenCache.aggregatedAttestations.isKnownCalls.inc();
+
+    for (let i = 0; i < seenAggregationInfoArr.length; i++) {
+      if (isSuperSetOrEqual(seenAggregationInfoArr[i].aggregationBits, aggregationBits)) {
+        this.metrics?.seenCache.aggregatedAttestations.superSetCheckTotal.observe(i + 1);
+        this.metrics?.seenCache.aggregatedAttestations.isKnownHits.inc();
+        return true;
+      }
+    }
+
+    this.metrics?.seenCache.aggregatedAttestations.superSetCheckTotal.observe(seenAggregationInfoArr.length);
+    return false;
+  }
+
+  add(targetEpoch: Epoch, attDataRoot: RootHex, newItem: AggregationInfo, checkIsKnown: boolean): void {
+    const {aggregationBits} = newItem;
+    if (checkIsKnown && this.isKnown(targetEpoch, attDataRoot, aggregationBits)) {
+      return;
+    }
+
+    const seenAggregationInfoArr = this.aggregateRootsByEpoch.getOrDefault(targetEpoch).getOrDefault(attDataRoot);
+    insertDesc(seenAggregationInfoArr, newItem);
+  }
+
+  prune(currentEpoch: Epoch): void {
+    this.lowestPermissibleEpoch = Math.max(currentEpoch - MAX_EPOCHS_IN_CACHE, 0);
+    for (const epoch of this.aggregateRootsByEpoch.keys()) {
+      if (epoch < this.lowestPermissibleEpoch) {
+        this.aggregateRootsByEpoch.delete(epoch);
+      }
+    }
+  }
+}
+
+/**
+ * Make sure seenAggregationInfoArr is always in desc order based on trueBitCount so that isKnown can be faster
+ */
+export function insertDesc(seenAggregationInfoArr: AggregationInfo[], newItem: AggregationInfo): void {
+  const {trueBitCount} = newItem;
+  let found = false;
+  for (let i = 0; i < seenAggregationInfoArr.length; i++) {
+    if (trueBitCount >= seenAggregationInfoArr[i].trueBitCount) {
+      seenAggregationInfoArr.splice(i, 0, newItem);
+      found = true;
+      break;
+    }
+  }
+
+  if (!found) seenAggregationInfoArr.push(newItem);
+}

packages/lodestar/src/chain/seenCache/seenCommitteeContribution.ts

@@ -1,5 +1,10 @@
 import {Slot, ValidatorIndex} from "@chainsafe/lodestar-types";
+import {ContributionAndProof, SyncCommitteeContribution} from "@chainsafe/lodestar-types/altair";
+import {toHexString} from "@chainsafe/ssz";
+import {IMetrics} from "../../metrics";
+import {isSuperSetOrEqual} from "../../util/bitArray";
 import {MapDef} from "../../util/map";
+import {AggregationInfo, insertDesc} from "./seenAggregateAndProof";
 
 /**
  * SyncCommittee aggregates are only useful for the next block they have signed.
@@ -9,38 +14,92 @@ const MAX_SLOTS_IN_CACHE = 8;
 /** AggregatorSubnetKey = `aggregatorIndex + subcommitteeIndex` */
 type AggregatorSubnetKey = string;
 
+/** ContributionDataKey = `slot + beacon_block_root + subcommittee_index */
+type ContributionDataKey = string;
+
 /**
  * Cache SyncCommitteeContribution and seen ContributionAndProof.
  * This is used for SignedContributionAndProof validation and block factory.
  * This stays in-memory and should be pruned per slot.
  */
 export class SeenContributionAndProof {
-  private readonly seenCacheBySlot = new MapDef<Slot, Set<AggregatorSubnetKey>>(() => new Set<AggregatorSubnetKey>());
+  private readonly seenAggregatorBySlot = new MapDef<Slot, Set<AggregatorSubnetKey>>(
+    () => new Set<AggregatorSubnetKey>()
+  );
+
+  private readonly seenContributionBySlot = new MapDef<Slot, MapDef<ContributionDataKey, AggregationInfo[]>>(
+    () => new MapDef(() => [])
+  );
+
+  constructor(private readonly metrics: IMetrics | null) {}
+
+  /**
+   * _[IGNORE]_ A valid sync committee contribution with equal `slot`, `beacon_block_root` and `subcommittee_index` whose
+   * `aggregation_bits` is non-strict superset has _not_ already been seen.
+   */
+  participantsKnown(contribution: SyncCommitteeContribution): boolean {
+    const {aggregationBits, slot} = contribution;
+    const contributionMap = this.seenContributionBySlot.getOrDefault(slot);
+    const seenAggregationInfoArr = contributionMap.getOrDefault(toContributionDataKey(contribution));
+    this.metrics?.seenCache.committeeContributions.isKnownCalls.inc();
+    // seenAttestingIndicesArr is sorted by trueBitCount desc
+
+    for (let i = 0; i < seenAggregationInfoArr.length; i++) {
+      if (isSuperSetOrEqual(seenAggregationInfoArr[i].aggregationBits, aggregationBits)) {
+        this.metrics?.seenCache.committeeContributions.isKnownHits.inc();
+        this.metrics?.seenCache.committeeContributions.superSetCheckTotal.observe(i + 1);
+        return true;
+      }
+    }
+
+    this.metrics?.seenCache.committeeContributions.superSetCheckTotal.observe(seenAggregationInfoArr.length);
+    return false;
+  }
 
   /**
    * Gossip validation requires to check:
    * The sync committee contribution is the first valid contribution received for the aggregator with index
    * contribution_and_proof.aggregator_index for the slot contribution.slot and subcommittee index contribution.subcommittee_index.
    */
-  isKnown(slot: Slot, subcommitteeIndex: number, aggregatorIndex: ValidatorIndex): boolean {
-    return this.seenCacheBySlot.get(slot)?.has(seenCacheKey(subcommitteeIndex, aggregatorIndex)) === true;
+  isAggregatorKnown(slot: Slot, subcommitteeIndex: number, aggregatorIndex: ValidatorIndex): boolean {
+    return this.seenAggregatorBySlot.get(slot)?.has(seenAggregatorKey(subcommitteeIndex, aggregatorIndex)) === true;
   }
 
   /** Register item as seen in the cache */
-  add(slot: Slot, subcommitteeIndex: number, aggregatorIndex: ValidatorIndex): void {
-    this.seenCacheBySlot.getOrDefault(slot).add(seenCacheKey(subcommitteeIndex, aggregatorIndex));
+  add(contributionAndProof: ContributionAndProof, trueBitCount: number): void {
+    const {contribution, aggregatorIndex} = contributionAndProof;
+    const {subcommitteeIndex, slot, aggregationBits} = contribution;
+
+    // add to seenAggregatorBySlot
+    this.seenAggregatorBySlot.getOrDefault(slot).add(seenAggregatorKey(subcommitteeIndex, aggregatorIndex));
+
+    // add to seenContributionBySlot
+    const contributionMap = this.seenContributionBySlot.getOrDefault(slot);
+    const seenAggregationInfoArr = contributionMap.getOrDefault(toContributionDataKey(contribution));
+    insertDesc(seenAggregationInfoArr, {aggregationBits, trueBitCount});
   }
 
   /** Prune per head slot */
   prune(headSlot: Slot): void {
-    for (const slot of this.seenCacheBySlot.keys()) {
+    for (const slot of this.seenAggregatorBySlot.keys()) {
       if (slot < headSlot - MAX_SLOTS_IN_CACHE) {
-        this.seenCacheBySlot.delete(slot);
+        this.seenAggregatorBySlot.delete(slot);
+      }
+    }
+
+    for (const slot of this.seenContributionBySlot.keys()) {
+      if (slot < headSlot - MAX_SLOTS_IN_CACHE) {
+        this.seenContributionBySlot.delete(slot);
       }
     }
   }
 }
 
-function seenCacheKey(subcommitteeIndex: number, aggregatorIndex: ValidatorIndex): AggregatorSubnetKey {
+function seenAggregatorKey(subcommitteeIndex: number, aggregatorIndex: ValidatorIndex): AggregatorSubnetKey {
   return `${subcommitteeIndex}-${aggregatorIndex}`;
 }
+
+function toContributionDataKey(contribution: SyncCommitteeContribution): ContributionDataKey {
+  const {slot, beaconBlockRoot, subcommitteeIndex} = contribution;
+  return `${slot} - ${toHexString(beaconBlockRoot)} - ${subcommitteeIndex}`;
+}