ChainSafe · g11tech · Feb 19, 2024 · Aug 8, 2023 · Aug 18, 2023 · Aug 18, 2023
@@ -871,6 +871,9 @@ export class BeaconChain implements IBeaconChain {
     metrics.forkChoice.balancesLength.set(forkChoiceMetrics.balancesLength);
     metrics.forkChoice.nodes.set(forkChoiceMetrics.nodes);
     metrics.forkChoice.indices.set(forkChoiceMetrics.indices);
+
+    const headState = this.getHeadState();
+    metrics.headState.unfinalizedPubkeyCacheSize.set(headState.epochCtx.unfinalizedPubkey2index.size);
   }
 
   private onClockSlot(slot: Slot): void {
@@ -959,6 +962,40 @@ export class BeaconChain implements IBeaconChain {
     if (headState) {
       this.opPool.pruneAll(headBlock, headState);
     }
+
+    const cpEpoch = cp.epoch;
+    const electraEpoch = headState?.config.ELECTRA_FORK_EPOCH ?? Infinity;
+
+    if (headState === null) {
+      this.logger.verbose("Head state is null");
+    } else if (cpEpoch >= electraEpoch) {
+      // Get the validator.length from the state at cpEpoch
+      // We are confident the last element in the list is from headEpoch
+      // Thus we query from the end of the list. (cpEpoch - headEpoch - 1) is negative number
+      const pivotValidatorIndex = headState.epochCtx.getValidatorCountAtEpoch(cpEpoch);
+
+      if (pivotValidatorIndex !== undefined) {
+        // Note EIP-6914 will break this logic
+        const newFinalizedValidators = headState.epochCtx.unfinalizedPubkey2index.filter(
+          (index, _pubkey) => index < pivotValidatorIndex
+        );
+
+        // Populate finalized pubkey cache and remove unfinalized pubkey cache
+        if (!newFinalizedValidators.isEmpty()) {
+          this.regen.updateUnfinalizedPubkeys(newFinalizedValidators);
+        }
+      }
+    }
+
+    // TODO-Electra: Deprecating eth1Data poll requires a check on a finalized checkpoint state.
+    // Will resolve this later
+    // if (cpEpoch >= (this.config.ELECTRA_FORK_EPOCH ?? Infinity)) {
+    //   // finalizedState can be safely casted to Electra state since cp is already post-Electra
+    //   if (finalizedState.eth1DepositIndex >= (finalizedState as CachedBeaconStateElectra).depositReceiptsStartIndex) {
+    //     // Signal eth1 to stop polling eth1Data
+    //     this.eth1.stopPollingEth1Data();
+    //   }
+    // }
   }
 
   async updateBeaconProposerData(epoch: Epoch, proposers: ProposerPreparationData[]): Promise<void> {

@@ -1,7 +1,7 @@
 import {toHexString} from "@chainsafe/ssz";
 import {phase0, Slot, allForks, RootHex, Epoch} from "@lodestar/types";
 import {IForkChoice, ProtoBlock} from "@lodestar/fork-choice";
-import {CachedBeaconStateAllForks, computeEpochAtSlot} from "@lodestar/state-transition";
+import {CachedBeaconStateAllForks, UnfinalizedPubkeyIndexMap, computeEpochAtSlot} from "@lodestar/state-transition";
 import {Logger} from "@lodestar/utils";
 import {routes} from "@lodestar/api";
 import {CheckpointHex, toCheckpointHex} from "../stateCache/index.js";
@@ -125,6 +125,54 @@ export class QueuedStateRegenerator implements IStateRegenerator {
     return this.checkpointStateCache.updatePreComputedCheckpoint(rootHex, epoch);
   }
 
+  /**
+   * Remove `validators` from all unfinalized cache's epochCtx.UnfinalizedPubkey2Index,
+   * and add them to epochCtx.pubkey2index and epochCtx.index2pubkey
+   */
+  updateUnfinalizedPubkeys(validators: UnfinalizedPubkeyIndexMap): void {
+    let numStatesUpdated = 0;
+    const states = this.stateCache.getStates();
+    const cpStates = this.checkpointStateCache.getStates();
+
+    // Add finalized pubkeys to all states.
+    const addTimer = this.metrics?.regenFnAddPubkeyTime.startTimer();
+
+    // We only need to add pubkeys to any one of the states since the finalized caches is shared globally across all states
+    const firstState = (states.next().value ?? cpStates.next().value) as CachedBeaconStateAllForks | undefined;
+
+    if (firstState !== undefined) {
+      firstState.epochCtx.addFinalizedPubkeys(validators, this.metrics?.epochCache ?? undefined);
+    } else {
+      this.logger.warn("Attempt to delete finalized pubkey from unfinalized pubkey cache. But no state is available");
+    }
+
+    addTimer?.();
+
+    // Delete finalized pubkeys from unfinalized pubkey cache for all states
+    const deleteTimer = this.metrics?.regenFnDeletePubkeyTime.startTimer();
+    const pubkeysToDelete = Array.from(validators.keys());
+
+    for (const s of states) {
+      s.epochCtx.deleteUnfinalizedPubkeys(pubkeysToDelete);
+      numStatesUpdated++;
+    }
+
+    for (const s of cpStates) {
+      s.epochCtx.deleteUnfinalizedPubkeys(pubkeysToDelete);
+      numStatesUpdated++;
+    }
+
+    // Since first state is consumed from the iterator. Will need to perform delete explicitly
+    if (firstState !== undefined) {
+      firstState?.epochCtx.deleteUnfinalizedPubkeys(pubkeysToDelete);
+      numStatesUpdated++;
+    }
+
+    deleteTimer?.();
+
+    this.metrics?.regenFnNumStatesUpdated.observe(numStatesUpdated);
+  }
+
   /**
    * Get the state to run with `block`.
    * - State after `block.parentRoot` dialed forward to block.slot

@@ -172,6 +172,10 @@ export class FIFOBlockStateCache implements BlockStateCache {
     }));
   }
 
+  getStates(): IterableIterator<CachedBeaconStateAllForks> {
+    throw new Error("Method not implemented.");
+  }
+
   /**
    * For unit test only.
    */

@@ -587,6 +587,10 @@ export class PersistentCheckpointStateCache implements CheckpointStateCache {
     });
   }
 
+  getStates(): IterableIterator<CachedBeaconStateAllForks> {
+    throw new Error("Method not implemented.");
+  }
+
   /** ONLY FOR DEBUGGING PURPOSES. For spec tests on error */
   dumpCheckpointKeys(): string[] {
     return Array.from(this.cache.keys());

@@ -33,7 +33,7 @@ export class StateContextCache implements BlockStateCache {
     this.maxStates = maxStates;
     this.cache = new MapTracker(metrics?.stateCache);
     if (metrics) {
-      this.metrics = metrics.stateCache;
+      this.metrics = {...metrics.stateCache, ...metrics.epochCache};
       metrics.stateCache.size.addCollect(() => metrics.stateCache.size.set(this.cache.size));
     }
   }
@@ -127,6 +127,10 @@ export class StateContextCache implements BlockStateCache {
     }));
   }
 
+  getStates(): IterableIterator<CachedBeaconStateAllForks> {
+    return this.cache.values();
+  }
+
   private deleteAllEpochItems(epoch: Epoch): void {
     for (const rootHex of this.epochIndex.get(epoch) || []) {
       this.cache.delete(rootHex);

@@ -167,6 +167,10 @@ export class CheckpointStateCache implements CheckpointStateCacheInterface {
     }));
   }
 
+  getStates(): IterableIterator<CachedBeaconStateAllForks> {
+    return this.cache.values();
+  }
+
   /** ONLY FOR DEBUGGING PURPOSES. For spec tests on error */
   dumpCheckpointKeys(): string[] {
     return Array.from(this.cache.keys());

@@ -28,6 +28,7 @@ export interface BlockStateCache {
   prune(headStateRootHex: RootHex): void;
   deleteAllBeforeEpoch(finalizedEpoch: Epoch): void;
   dumpSummary(): routes.lodestar.StateCacheItem[];
+  getStates(): IterableIterator<CachedBeaconStateAllForks>; // Expose beacon states stored in cache. Use with caution
 }
 
 /**
@@ -65,6 +66,7 @@ export interface CheckpointStateCache {
   processState(blockRootHex: RootHex, state: CachedBeaconStateAllForks): Promise<number>;
   clear(): void;
   dumpSummary(): routes.lodestar.StateCacheItem[];
+  getStates(): IterableIterator<CachedBeaconStateAllForks>; // Expose beacon states stored in cache. Use with caution
 }
 
 export enum CacheItemType {

@@ -1,6 +1,11 @@
 import {phase0, ssz} from "@lodestar/types";
 import {ChainForkConfig} from "@lodestar/config";
-import {BeaconStateAllForks, becomesNewEth1Data} from "@lodestar/state-transition";
+import {
+  BeaconStateAllForks,
+  CachedBeaconStateAllForks,
+  CachedBeaconStateElectra,
+  becomesNewEth1Data,
+} from "@lodestar/state-transition";
 import {ErrorAborted, TimeoutError, fromHex, Logger, isErrorAborted, sleep} from "@lodestar/utils";
 
 import {IBeaconDb} from "../db/index.js";
@@ -67,6 +72,8 @@ export class Eth1DepositDataTracker {
   /** Dynamically adjusted batch size to fetch deposit logs */
   private eth1GetLogsBatchSizeDynamic = MAX_BLOCKS_PER_LOG_QUERY;
   private readonly forcedEth1DataVote: phase0.Eth1Data | null;
+  /** To stop `runAutoUpdate()` in addition to AbortSignal */
+  private stopPolling: boolean;
 
   constructor(
     opts: Eth1Options,
@@ -81,6 +88,8 @@ export class Eth1DepositDataTracker {
     this.depositsCache = new Eth1DepositsCache(opts, config, db);
     this.eth1DataCache = new Eth1DataCache(config, db);
     this.eth1FollowDistance = config.ETH1_FOLLOW_DISTANCE;
+    // TODO Electra: fix scenario where node starts post-Electra and `stopPolling` will always be false
+    this.stopPolling = false;
 
     this.forcedEth1DataVote = opts.forcedEth1DataVote
       ? ssz.phase0.Eth1Data.deserialize(fromHex(opts.forcedEth1DataVote))
@@ -109,10 +118,22 @@ export class Eth1DepositDataTracker {
     }
   }
 
+  // TODO Electra: Figure out how an elegant way to stop eth1data polling
+  stopPollingEth1Data(): void {
+    this.stopPolling = true;
+  }
+
   /**
    * Return eth1Data and deposits ready for block production for a given state
    */
-  async getEth1DataAndDeposits(state: BeaconStateAllForks): Promise<Eth1DataAndDeposits> {
+  async getEth1DataAndDeposits(state: CachedBeaconStateAllForks): Promise<Eth1DataAndDeposits> {
+    if (
+      state.epochCtx.isAfterElectra() &&
+      state.eth1DepositIndex >= (state as CachedBeaconStateElectra).depositReceiptsStartIndex
+    ) {
+      // No need to poll eth1Data since Electra deprecates the mechanism after depositReceiptsStartIndex is reached
+      return {eth1Data: state.eth1Data, deposits: []};
+    }
     const eth1Data = this.forcedEth1DataVote ?? (await this.getEth1Data(state));
     const deposits = await this.getDeposits(state, eth1Data);
     return {eth1Data, deposits};
@@ -141,7 +162,10 @@ export class Eth1DepositDataTracker {
    * Returns deposits to be included for a given state and eth1Data vote.
    * Requires internal caches to be updated regularly to return good results
    */
-  private async getDeposits(state: BeaconStateAllForks, eth1DataVote: phase0.Eth1Data): Promise<phase0.Deposit[]> {
+  private async getDeposits(
+    state: CachedBeaconStateAllForks,
+    eth1DataVote: phase0.Eth1Data
+  ): Promise<phase0.Deposit[]> {
     // No new deposits have to be included, continue
     if (eth1DataVote.depositCount === state.eth1DepositIndex) {
       return [];
@@ -162,7 +186,7 @@ export class Eth1DepositDataTracker {
   private async runAutoUpdate(): Promise<void> {
     let lastRunMs = 0;
 
-    while (!this.signal.aborted) {
+    while (!this.signal.aborted && !this.stopPolling) {
       lastRunMs = Date.now();
 
       try {

@@ -106,6 +106,10 @@ export class Eth1ForBlockProduction implements IEth1ForBlockProduction {
   startPollingMergeBlock(): void {
     return this.eth1MergeBlockTracker.startPollingMergeBlock();
   }
+
+  stopPollingEth1Data(): void {
+    return this.eth1DepositDataTracker?.stopPollingEth1Data();
+  }
 }
 
 /**
@@ -140,4 +144,8 @@ export class Eth1ForBlockProductionDisabled implements IEth1ForBlockProduction {
   startPollingMergeBlock(): void {
     // Ignore
   }
+
+  stopPollingEth1Data(): void {
+    // Ignore
+  }
 }
@@ -62,6 +62,11 @@ export interface IEth1ForBlockProduction {
    * - head state not isMergeTransitionComplete
    */
   startPollingMergeBlock(): void;
+
+  /**
+   * Should stop polling eth1Data after a Electra block is finalized AND deposit_receipts_start_index is reached
+   */
+  stopPollingEth1Data(): void;
 }
 
 /** Different Eth1Block from phase0.Eth1Block with blockHash */

@@ -1,17 +1,17 @@
 import {toGindex, Tree} from "@chainsafe/persistent-merkle-tree";
 import {toHexString} from "@chainsafe/ssz";
-import {MAX_DEPOSITS} from "@lodestar/params";
-import {BeaconStateAllForks} from "@lodestar/state-transition";
+import {CachedBeaconStateAllForks} from "@lodestar/state-transition";
 import {phase0, ssz} from "@lodestar/types";
 import {FilterOptions} from "@lodestar/db";
+import {getEth1DepositCount} from "@lodestar/state-transition";
 import {Eth1Error, Eth1ErrorCode} from "../errors.js";
 import {DepositTree} from "../../db/repositories/depositDataRoot.js";
 
 export type DepositGetter<T> = (indexRange: FilterOptions<number>, eth1Data: phase0.Eth1Data) => Promise<T[]>;
 
 export async function getDeposits<T>(
   // eth1_deposit_index represents the next deposit index to be added
-  state: BeaconStateAllForks,
+  state: CachedBeaconStateAllForks,
   eth1Data: phase0.Eth1Data,
   depositsGetter: DepositGetter<T>
 ): Promise<T[]> {
@@ -22,9 +22,11 @@ export async function getDeposits<T>(
     throw new Eth1Error({code: Eth1ErrorCode.DEPOSIT_INDEX_TOO_HIGH, depositIndex, depositCount});
   }
 
-  // Spec v0.12.2
-  // assert len(body.deposits) == min(MAX_DEPOSITS, state.eth1_data.deposit_count - state.eth1_deposit_index)
-  const depositsLen = Math.min(MAX_DEPOSITS, depositCount - depositIndex);
+  const depositsLen = getEth1DepositCount(state, eth1Data);
+
+  if (depositsLen === 0) {
+    return []; // If depositsLen === 0, we can return early since no deposit with be returned from depositsGetter
+  }
 
   const indexRange = {gte: depositIndex, lt: depositIndex + depositsLen};
   const deposits = await depositsGetter(indexRange, eth1Data);

@@ -176,7 +176,9 @@ export class ExecutionEngineHttp implements IExecutionEngine {
     parentBlockRoot?: Root
   ): Promise<ExecutePayloadResponse> {
     const method =
-      ForkSeq[fork] >= ForkSeq.deneb
+      ForkSeq[fork] >= ForkSeq.electra
+        ? "engine_newPayloadV6110"
+        : ForkSeq[fork] >= ForkSeq.deneb
         ? "engine_newPayloadV3"
         : ForkSeq[fork] >= ForkSeq.capella
         ? "engine_newPayloadV2"
@@ -196,7 +198,7 @@ export class ExecutionEngineHttp implements IExecutionEngine {
       const serializedVersionedHashes = serializeVersionedHashes(versionedHashes);
       const parentBeaconBlockRoot = serializeBeaconBlockRoot(parentBlockRoot);
 
-      const method = "engine_newPayloadV3";
+      const method = ForkSeq[fork] >= ForkSeq.electra ? "engine_newPayloadV6110" : "engine_newPayloadV3";
       engineRequest = {
         method,
         params: [serializedExecutionPayload, serializedVersionedHashes, parentBeaconBlockRoot],
@@ -370,7 +372,9 @@ export class ExecutionEngineHttp implements IExecutionEngine {
     shouldOverrideBuilder?: boolean;
   }> {
     const method =
-      ForkSeq[fork] >= ForkSeq.deneb
+      ForkSeq[fork] >= ForkSeq.electra
+        ? "engine_getPayloadV6110"
+        : ForkSeq[fork] >= ForkSeq.deneb
         ? "engine_getPayloadV3"
         : ForkSeq[fork] >= ForkSeq.capella
         ? "engine_getPayloadV2"

@@ -3,10 +3,10 @@ import {KZGCommitment, Blob, KZGProof} from "@lodestar/types/deneb";
 import {Root, RootHex, allForks, capella, Wei} from "@lodestar/types";
 
 import {DATA} from "../../eth1/provider/utils.js";
-import {PayloadIdCache, PayloadId, WithdrawalV1} from "./payloadIdCache.js";
+import {PayloadIdCache, PayloadId, WithdrawalV1, DepositReceiptV1} from "./payloadIdCache.js";
 import {ExecutionPayloadBody} from "./types.js";
 
-export {PayloadIdCache, type PayloadId, type WithdrawalV1};
+export {PayloadIdCache, type PayloadId, type WithdrawalV1, type DepositReceiptV1};
 
 export enum ExecutionPayloadStatus {
   /** given payload is valid */