fix: aggregated attestation pool for electra (#7656)

**Motivation**

- we have 2 bugs for electra logic:
  - one logic in phase0 cannot be applied to electra
  - wrong for loop place
- one bug for all forks: incorrect dependent root used when checking
attestation data, see #7651

**Description**

extract fixes from #7646
- add more metrics for the pool
- remove this logic for electra: "after 2 slots, there are a good chance
that we have 2 * MAX_ATTESTATIONS_ELECTRA attestations and break the for
loop early"
- correct the for loop place to limit attestation consolidations
- fix `isValidShuffling()` for #7651 
- unit tests are done in #7646 , since logic changes there I cannot
bring them here

---------

Co-authored-by: Tuyen Nguyen <[email protected]>
Co-authored-by: Nico Flaig <[email protected]>

Author: twoeths

packages/beacon-node/src/chain/chain.ts

@@ -239,7 +239,7 @@ export class BeaconChain implements IBeaconChain {
       preAggregateCutOffTime,
       this.opts?.preaggregateSlotDistance
     );
-    this.aggregatedAttestationPool = new AggregatedAttestationPool(this.config);
+    this.aggregatedAttestationPool = new AggregatedAttestationPool(this.config, metrics);
     this.syncCommitteeMessagePool = new SyncCommitteeMessagePool(
       clock,
       preAggregateCutOffTime,
@@ -373,7 +373,7 @@ export class BeaconChain implements IBeaconChain {
     }
 
     if (metrics) {
-      metrics.opPool.aggregatedAttestationPoolSize.addCollect(() => this.onScrapeMetrics(metrics));
+      metrics.opPool.attestationPoolSize.addCollect(() => this.onScrapeMetrics(metrics));
     }
 
     // Event handlers. emitter is created internally and dropped on close(). Not need to .removeListener()
@@ -1076,9 +1076,7 @@ export class BeaconChain implements IBeaconChain {
   }
 
   private onScrapeMetrics(metrics: Metrics): void {
-    const {attestationCount, attestationDataCount} = this.aggregatedAttestationPool.getAttestationCount();
-    metrics.opPool.aggregatedAttestationPoolSize.set(attestationCount);
-    metrics.opPool.aggregatedAttestationPoolUniqueData.set(attestationDataCount);
+    // aggregatedAttestationPool tracks metrics on its own
     metrics.opPool.attestationPoolSize.set(this.attestationPool.getAttestationCount());
     metrics.opPool.attesterSlashingPoolSize.set(this.opPool.attesterSlashingsSize);
     metrics.opPool.proposerSlashingPoolSize.set(this.opPool.proposerSlashingsSize);

packages/beacon-node/src/chain/opPools/aggregatedAttestationPool.ts

@@ -34,7 +34,9 @@ import {
   ssz,
 } from "@lodestar/types";
 import {assert, MapDef, toRootHex} from "@lodestar/utils";
+import {Metrics} from "../../metrics/metrics.js";
 import {IntersectResult, intersectUint8Arrays} from "../../util/bitArray.js";
+import {getShufflingDependentRoot} from "../../util/dependentRoot.js";
 import {InsertOutcome} from "./types.js";
 import {pruneBySlot, signatureFromBytesNoCheck} from "./utils.js";
 
@@ -104,21 +106,11 @@ export class AggregatedAttestationPool {
   >(() => new Map<DataRootHex, Map<CommitteeIndex, MatchingDataAttestationGroup>>());
   private lowestPermissibleSlot = 0;
 
-  constructor(private readonly config: ChainForkConfig) {}
-
-  /** For metrics to track size of the pool */
-  getAttestationCount(): {attestationCount: number; attestationDataCount: number} {
-    let attestationCount = 0;
-    let attestationDataCount = 0;
-    for (const attestationGroupByIndexByDataHex of this.attestationGroupByIndexByDataHexBySlot.values()) {
-      for (const attestationGroupByIndex of attestationGroupByIndexByDataHex.values()) {
-        attestationDataCount += attestationGroupByIndex.size;
-        for (const attestationGroup of attestationGroupByIndex.values()) {
-          attestationCount += attestationGroup.getAttestationCount();
-        }
-      }
-    }
-    return {attestationCount, attestationDataCount};
+  constructor(
+    private readonly config: ChainForkConfig,
+    private readonly metrics: Metrics | null = null
+  ) {
+    metrics?.opPool.aggregatedAttestationPool.attDataPerSlot.addCollect(() => this.onScrapeMetrics(metrics));
   }
 
   add(
@@ -317,10 +309,7 @@ export class AggregatedAttestationPool {
     const slots = Array.from(this.attestationGroupByIndexByDataHexBySlot.keys()).sort((a, b) => b - a);
     // Track score of each `AttestationsConsolidation`
     const consolidations = new Map<AttestationsConsolidation, number>();
-    let minScore = Number.MAX_SAFE_INTEGER;
-    let slotCount = 0;
     slot: for (const slot of slots) {
-      slotCount++;
       const attestationGroupByIndexByDataHash = this.attestationGroupByIndexByDataHexBySlot.get(slot);
       // should not happen
       if (!attestationGroupByIndexByDataHash) {
@@ -338,34 +327,30 @@ export class AggregatedAttestationPool {
       }
 
       const slotDelta = stateSlot - slot;
-      // CommitteeIndex    0           1            2    ...   Consolidation
+      // CommitteeIndex    0           1            2    ...   Consolidationn (sameAttDataCons)
       // Attestations    att00  ---   att10  ---  att20  ---   0 (att 00 10 20)
       //                 att01  ---     -    ---  att21  ---   1 (att 01 __ 21)
       //                   -    ---     -    ---  att22  ---   2 (att __ __ 22)
       for (const attestationGroupByIndex of attestationGroupByIndexByDataHash.values()) {
         // sameAttDataCons could be up to MAX_ATTESTATIONS_PER_GROUP_ELECTRA
         const sameAttDataCons: AttestationsConsolidation[] = [];
+        const allAttestationGroups = Array.from(attestationGroupByIndex.values());
+        if (allAttestationGroups.length === 0) {
+          continue;
+        }
+
+        if (!validateAttestationDataFn(allAttestationGroups[0].data)) {
+          continue;
+        }
+
         for (const [committeeIndex, attestationGroup] of attestationGroupByIndex.entries()) {
           const notSeenAttestingIndices = notSeenValidatorsFn(epoch, slot, committeeIndex);
           if (notSeenAttestingIndices === null || notSeenAttestingIndices.size === 0) {
             continue;
           }
 
-          if (
-            slotCount > 2 &&
-            consolidations.size >= MAX_ATTESTATIONS_ELECTRA &&
-            notSeenAttestingIndices.size / slotDelta < minScore
-          ) {
-            // after 2 slots, there are a good chance that we have 2 * MAX_ATTESTATIONS_ELECTRA attestations and break the for loop early
-            // if not, we may have to scan all slots in the pool
-            // if we have enough attestations and the max possible score is lower than scores of `attestationsByScore`, we should skip
-            // otherwise it takes time to check attestation, add it and remove it later after the sort by score
-            continue;
-          }
-
-          if (!validateAttestationDataFn(attestationGroup.data)) {
-            continue;
-          }
+          // cannot apply this optimization like pre-electra because consolidation needs to be done across committees:
+          // "after 2 slots, there are a good chance that we have 2 * MAX_ATTESTATIONS_ELECTRA attestations and break the for loop early"
 
           // TODO: Is it necessary to validateAttestation for:
           // - Attestation committee index not within current committee count
@@ -387,18 +372,16 @@ export class AggregatedAttestationPool {
             sameAttDataCons[i].byCommittee.set(committeeIndex, attestationNonParticipation);
             sameAttDataCons[i].totalNotSeenCount += attestationNonParticipation.notSeenAttesterCount;
           }
-          for (const consolidation of sameAttDataCons) {
-            const score = consolidation.totalNotSeenCount / slotDelta;
-            if (score < minScore) {
-              minScore = score;
-            }
+        } // all committees are processed
 
-            consolidations.set(consolidation, score);
+        // after all committees are processed, we have a list of sameAttDataCons
+        for (const consolidation of sameAttDataCons) {
+          const score = consolidation.totalNotSeenCount / slotDelta;
+          consolidations.set(consolidation, score);
 
-            // Stop accumulating attestations there are enough that may have good scoring
-            if (consolidations.size >= MAX_ATTESTATIONS_ELECTRA * 2) {
-              break slot;
-            }
+          // Stop accumulating attestations there are enough that may have good scoring
+          if (consolidations.size >= MAX_ATTESTATIONS_ELECTRA * 2) {
+            break slot;
           }
         }
       }
@@ -440,6 +423,53 @@ export class AggregatedAttestationPool {
     }
     return attestations;
   }
+
+  private onScrapeMetrics({opPool}: Metrics): void {
+    const metrics = opPool.aggregatedAttestationPool;
+    const allSlots = Array.from(this.attestationGroupByIndexByDataHexBySlot.keys());
+
+    // always record the previous slot because the current slot may not be finished yet, we may receive more attestations
+    if (allSlots.length > 1) {
+      // last item is current slot, we want the previous one
+      const previousSlot = allSlots.at(-2);
+      if (previousSlot == null) {
+        // only happen right after we start the node
+        return;
+      }
+
+      const groupByIndexByDataHex = this.attestationGroupByIndexByDataHexBySlot.get(previousSlot);
+      if (groupByIndexByDataHex != null) {
+        metrics.attDataPerSlot.set(groupByIndexByDataHex.size);
+
+        let maxAttestations = 0;
+        let committeeCount = 0;
+        for (const groupByIndex of groupByIndexByDataHex.values()) {
+          for (const group of groupByIndex.values()) {
+            const attestationCount = group.getAttestationCount();
+            maxAttestations = Math.max(maxAttestations, attestationCount);
+            metrics.attestationsPerCommittee.observe(attestationCount);
+            committeeCount += 1;
+          }
+        }
+        metrics.maxAttestationsPerCommittee.set(maxAttestations);
+        metrics.committeesPerSlot.set(committeeCount);
+      }
+    }
+
+    let attestationCount = 0;
+    let attestationDataCount = 0;
+    for (const attestationGroupByIndexByDataHex of this.attestationGroupByIndexByDataHexBySlot.values()) {
+      for (const attestationGroupByIndex of attestationGroupByIndexByDataHex.values()) {
+        attestationDataCount += attestationGroupByIndex.size;
+        for (const attestationGroup of attestationGroupByIndex.values()) {
+          attestationCount += attestationGroup.getAttestationCount();
+        }
+      }
+    }
+
+    metrics.poolSize.set(attestationCount);
+    metrics.poolUniqueData.set(attestationDataCount);
+  }
 }
 
 interface AttestationWithIndex {
@@ -815,7 +845,14 @@ function isValidShuffling(
 
   let attestationDependentRoot: string;
   try {
-    attestationDependentRoot = forkChoice.getDependentRoot(beaconBlock, EpochDifference.previous);
+    // should not use forkChoice.getDependentRoot directly, see https://github.com/ChainSafe/lodestar/issues/7651
+    // attestationDependentRoot = forkChoice.getDependentRoot(beaconBlock, EpochDifference.previous);
+    attestationDependentRoot = getShufflingDependentRoot(
+      forkChoice,
+      targetEpoch,
+      computeEpochAtSlot(beaconBlock.slot),
+      beaconBlock
+    );
   } catch (_) {
     // getDependent root may throw error if the dependent root of attestation data is prior to finalized slot
     // ignore this attestation data in that case since we're not sure it's compatible to the state

packages/beacon-node/src/metrics/metrics/lodestar.ts

@@ -862,17 +862,34 @@ export function createLodestarMetrics(
     },
 
     opPool: {
-      // Note: Current opPool metrics only track current size.
-      //       I don't believe tracking total add() count is relevant since that can be seen with gossip ACCEPTs
-      aggregatedAttestationPoolSize: register.gauge({
-        name: "lodestar_oppool_aggregated_attestation_pool_size",
-        help: "Current size of the AggregatedAttestationPool = total attestations",
-      }),
-      /** This metric helps view how many overlapping attestations we keep per data on average */
-      aggregatedAttestationPoolUniqueData: register.gauge({
-        name: "lodestar_oppool_aggregated_attestation_pool_unique_data_count",
-        help: "Current size of the AggregatedAttestationPool = total attestations unique by data",
-      }),
+      aggregatedAttestationPool: {
+        poolSize: register.gauge({
+          name: "lodestar_oppool_aggregated_attestation_pool_size",
+          help: "Current size of the AggregatedAttestationPool = total attestations",
+        }),
+        poolUniqueData: register.gauge({
+          name: "lodestar_oppool_aggregated_attestation_pool_unique_data_count",
+          help: "Current size of the AggregatedAttestationPool = total attestations unique by data",
+        }),
+        attDataPerSlot: register.gauge({
+          name: "lodestar_oppool_aggregated_attestation_pool_attestation_data_per_slot_total",
+          help: "Total number of attestation data per slot in AggregatedAttestationPool",
+        }),
+        committeesPerSlot: register.gauge({
+          name: "lodestar_oppool_aggregated_attestation_pool_committees_per_slot_total",
+          help: "Total number of committees per slot in AggregatedAttestationPool",
+        }),
+        // max number of attestations per committee will become number of consolidations
+        maxAttestationsPerCommittee: register.gauge({
+          name: "lodestar_oppool_aggregated_attestation_pool_max_attestations_per_committee",
+          help: "Max number of attestations per committee in AggregatedAttestationPool",
+        }),
+        attestationsPerCommittee: register.histogram({
+          name: "lodestar_oppool_aggregated_attestation_pool_attestations_per_committee",
+          help: "Number of attestations per committee in AggregatedAttestationPool",
+          buckets: [0, 2, 4, 8],
+        }),
+      },
       attestationPoolSize: register.gauge({
         name: "lodestar_oppool_attestation_pool_size",
         help: "Current size of the AttestationPool = total attestations unique by data and slot",

packages/beacon-node/test/unit/chain/opPools/aggregatedAttestationPool.test.ts

@@ -126,7 +126,7 @@ describe("AggregatedAttestationPool - Altair", () => {
         aggregationBits.getTrueBitIndexes().length,
         committee
       );
-      forkchoiceStub.getBlockHex.mockReturnValue(generateProtoBlock());
+      forkchoiceStub.getBlockHex.mockReturnValue(generateProtoBlock({slot: attestation.data.slot}));
       forkchoiceStub.getDependentRoot.mockReturnValue(ZERO_HASH_HEX);
       if (isReturned) {
         expect(pool.getAttestationsForBlock(fork, forkchoiceStub, altairState).length).toBeGreaterThan(0);
@@ -152,7 +152,7 @@ describe("AggregatedAttestationPool - Altair", () => {
     // all attesters are not seen
     const attestingIndices = [2, 3];
     pool.add(attestation, attDataRootHex, attestingIndices.length, committee);
-    forkchoiceStub.getBlockHex.mockReturnValue(generateProtoBlock());
+    forkchoiceStub.getBlockHex.mockReturnValue(generateProtoBlock({slot: attestation.data.slot}));
     forkchoiceStub.getDependentRoot.mockReturnValue("0xWeird");
     expect(pool.getAttestationsForBlock(fork, forkchoiceStub, altairState)).toEqual([]);
     // "forkchoice should be called to check pivot block"