Issue advance fcU for builing the EL block

rebaseing to the refactored prepare beacon proposer

refac payload id cache as separate class and add pruning

issue payload fcus if synced

rename issueNext.. to maybeIssueNext...

packages/lodestar/src/chain/beaconProposerCache.ts

@@ -30,7 +30,11 @@ export class BeaconProposerCache {
     }
   }
 
-  get(proposerIndex: number | string): string {
+  getOrDefault(proposerIndex: number | string): string {
     return this.feeRecipientByValidatorIndex.getOrDefault(`${proposerIndex}`).feeRecipient;
   }
+
+  get(proposerIndex: number | string): string | undefined {
+    return this.feeRecipientByValidatorIndex.get(`${proposerIndex}`)?.feeRecipient;
+  }
 }

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

@@ -1,20 +1,21 @@
 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";
 import {
   CachedBeaconStateAllForks,
   CachedBeaconStateAltair,
   computeStartSlotAtEpoch,
   getEffectiveBalanceIncrementsZeroInactive,
   altair,
   computeEpochAtSlot,
+  bellatrix,
+  allForks,
 } from "@chainsafe/lodestar-beacon-state-transition";
 import {IForkChoice, OnBlockPrecachedData, ForkChoiceError, ForkChoiceErrorCode} from "@chainsafe/lodestar-fork-choice";
 import {ILogger} from "@chainsafe/lodestar-utils";
 import {IChainForkConfig} from "@chainsafe/lodestar-config";
 import {IMetrics} from "../../metrics";
-import {IExecutionEngine} from "../../executionEngine";
+import {IExecutionEngine, PayloadId} from "../../executionEngine/interface";
 import {IBeaconDb} from "../../db";
 import {ZERO_HASH_HEX} from "../../constants";
 import {CheckpointStateCache, StateContextCache, toCheckpointHex} from "../stateCache";
@@ -25,6 +26,10 @@ import {getCheckpointFromState} from "./utils/checkpoint";
 import {PendingEvents} from "./utils/pendingEvents";
 import {FullyVerifiedBlock} from "./types";
 import {SeenAggregatedAttestations} from "../seenCache/seenAggregateAndProof";
+import {prepareExecutionPayload} from "../factory/block/body";
+import {IEth1ForBlockProduction} from "../../eth1";
+import {BeaconProposerCache} from "../beaconProposerCache";
+import {IBeaconClock} from "../clock";
 
 /**
  * Fork-choice allows to import attestations from current (0) or past (1) epoch.
@@ -37,10 +42,13 @@ export type ImportBlockModules = {
   stateCache: StateContextCache;
   checkpointStateCache: CheckpointStateCache;
   seenAggregatedAttestations: SeenAggregatedAttestations;
+  beaconProposerCache: BeaconProposerCache;
   lightClientServer: LightClientServer;
+  eth1: IEth1ForBlockProduction;
   executionEngine: IExecutionEngine;
   emitter: ChainEventEmitter;
   config: IChainForkConfig;
+  clock: IBeaconClock;
   logger: ILogger;
   metrics: IMetrics | null;
 };
@@ -180,6 +188,8 @@ export async function importBlock(chain: ImportBlockModules, fullyVerifiedBlock:
   const oldHead = chain.forkChoice.getHead();
   chain.forkChoice.updateHead();
   const newHead = chain.forkChoice.getHead();
+  const currFinalizedEpoch = chain.forkChoice.getFinalizedCheckpoint().epoch;
+
   if (newHead.blockRoot !== oldHead.blockRoot) {
     // new head
     pendingEvents.push(ChainEvent.forkChoiceHead, newHead);
@@ -208,30 +218,33 @@ export async function importBlock(chain: ImportBlockModules, fullyVerifiedBlock:
     }
   }
 
-  // NOTE: forkChoice.fsStore.finalizedCheckpoint MUST only change is response to an onBlock event
-  // Notify execution layer of head and finalized updates
-  const currFinalizedEpoch = chain.forkChoice.getFinalizedCheckpoint().epoch;
-  if (newHead.blockRoot !== oldHead.blockRoot || currFinalizedEpoch !== prevFinalizedEpoch) {
-    /**
-     * On post BELLATRIX_EPOCH but pre TTD, blocks include empty execution payload with a zero block hash.
-     * The consensus clients must not send notifyForkchoiceUpdate before TTD since the execution client will error.
-     * So we must check that:
-     * - `headBlockHash !== null` -> Pre BELLATRIX_EPOCH
-     * - `headBlockHash !== ZERO_HASH` -> Pre TTD
-     */
-    const headBlockHash = chain.forkChoice.getHead().executionPayloadBlockHash;
-    /**
-     * After BELLATRIX_EPOCH and TTD it's okay to send a zero hash block hash for the finalized block. This will happen if
-     * the current finalized block does not contain any execution payload at all (pre MERGE_EPOCH) or if it contains a
-     * zero block hash (pre TTD)
-     */
-    const finalizedBlockHash = chain.forkChoice.getFinalizedBlock().executionPayloadBlockHash;
-    if (headBlockHash !== null && headBlockHash !== ZERO_HASH_HEX) {
-      chain.executionEngine.notifyForkchoiceUpdate(headBlockHash, finalizedBlockHash ?? ZERO_HASH_HEX).catch((e) => {
-        chain.logger.error("Error pushing notifyForkchoiceUpdate()", {headBlockHash, finalizedBlockHash}, e);
-      });
+  void maybeIssueNextProposerEngineFcU(chain, postState).then((payloadId) => {
+    // NOTE: forkChoice.fsStore.finalizedCheckpoint MUST only change is response to an onBlock event
+    // Notify execution layer of head and finalized updates only if has already
+    // not been done via payloadId generation. But even if this fcU follows the
+    // payloadId one, there is no harm as the ELs will just ignore it.
+    if (payloadId === null && (newHead.blockRoot !== oldHead.blockRoot || currFinalizedEpoch !== prevFinalizedEpoch)) {
+      /**
+       * On post BELLATRIX_EPOCH but pre TTD, blocks include empty execution payload with a zero block hash.
+       * The consensus clients must not send notifyForkchoiceUpdate before TTD since the execution client will error.
+       * So we must check that:
+       * - `headBlockHash !== null` -> Pre BELLATRIX_EPOCH
+       * - `headBlockHash !== ZERO_HASH` -> Pre TTD
+       */
+      const headBlockHash = chain.forkChoice.getHead().executionPayloadBlockHash ?? ZERO_HASH_HEX;
+      /**
+       * After BELLATRIX_EPOCH and TTD it's okay to send a zero hash block hash for the finalized block. This will happen if
+       * the current finalized block does not contain any execution payload at all (pre MERGE_EPOCH) or if it contains a
+       * zero block hash (pre TTD)
+       */
+      const finalizedBlockHash = chain.forkChoice.getFinalizedBlock().executionPayloadBlockHash ?? ZERO_HASH_HEX;
+      if (headBlockHash !== ZERO_HASH_HEX) {
+        chain.executionEngine.notifyForkchoiceUpdate(headBlockHash, finalizedBlockHash).catch((e) => {
+          chain.logger.error("Error pushing notifyForkchoiceUpdate()", {headBlockHash, finalizedBlockHash}, e);
+        });
+      }
     }
-  }
+  });
 
   // Emit ChainEvent.block event
   //
@@ -250,6 +263,36 @@ export async function importBlock(chain: ImportBlockModules, fullyVerifiedBlock:
   pendingEvents.emit();
 }
 
+async function maybeIssueNextProposerEngineFcU(
+  chain: ImportBlockModules,
+  state: CachedBeaconStateAllForks
+): Promise<PayloadId | null> {
+  const prepareSlot = state.slot + 1;
+  // No need to try building block if we are not synced
+  if (prepareSlot > chain.clock.currentSlot + 1) {
+    return null;
+  }
+  const prepareState = allForks.processSlots(state, prepareSlot);
+  // TODO wait till third/last interval of the slot to actual send an fcU
+  // so that any head change is accomodated before that. However this could
+  // be optimized if the last block receieved is already head. This will be
+  // especially meaningful for mev boost which might have more delays
+  // because of how protocol is designed
+  if (bellatrix.isBellatrixStateType(prepareState)) {
+    try {
+      const proposerIndex = prepareState.epochCtx.getBeaconProposer(prepareSlot);
+      const feeRecipient = chain.beaconProposerCache.get(proposerIndex);
+      if (feeRecipient) {
+        const finalizedBlockHash = chain.forkChoice.getFinalizedBlock().executionPayloadBlockHash ?? ZERO_HASH_HEX;
+        return prepareExecutionPayload(chain, finalizedBlockHash, prepareState, feeRecipient);
+      }
+    } catch (e) {
+      chain.logger.error("Error on issuing next proposer engine fcU", {}, e as Error);
+    }
+  }
+  return null;
+}
+
 /**
  * Returns the closest state to postState.currentJustifiedCheckpoint in the same fork as postState
  *

packages/lodestar/src/chain/chain.ts

@@ -199,6 +199,7 @@ export class BeaconChain implements IBeaconChain {
         stateCache,
         checkpointStateCache,
         seenAggregatedAttestations: this.seenAggregatedAttestations,
+        beaconProposerCache: this.beaconProposerCache,
         emitter,
         config,
         logger,

packages/lodestar/src/chain/factory/block/body.ts

@@ -23,10 +23,14 @@ import {
   getCurrentEpoch,
   bellatrix,
 } from "@chainsafe/lodestar-beacon-state-transition";
+import {IChainForkConfig} from "@chainsafe/lodestar-config";
+import {toHex} from "@chainsafe/lodestar-utils";
 
 import {IBeaconChain} from "../../interface";
-import {PayloadId} from "../../../executionEngine/interface";
+import {PayloadId, IExecutionEngine} from "../../../executionEngine/interface";
 import {ZERO_HASH, ZERO_HASH_HEX} from "../../../constants";
+import {IEth1ForBlockProduction} from "../../../eth1";
+import {numToQuantity} from "../../../eth1/provider/utils";
 
 export async function assembleBody(
   chain: IBeaconChain,
@@ -89,7 +93,7 @@ export async function assembleBody(
     // - Call prepareExecutionPayload again if parameters change
 
     const finalizedBlockHash = chain.forkChoice.getFinalizedBlock().executionPayloadBlockHash;
-    const feeRecipient = chain.beaconProposerCache.get(proposerIndex);
+    const feeRecipient = chain.beaconProposerCache.getOrDefault(proposerIndex);
 
     // prepareExecutionPayload will throw error via notifyForkchoiceUpdate if
     // the EL returns Syncing on this request to prepare a payload
@@ -129,8 +133,12 @@ export async function assembleBody(
  *
  * @returns PayloadId = pow block found, null = pow NOT found
  */
-async function prepareExecutionPayload(
-  chain: IBeaconChain,
+export async function prepareExecutionPayload(
+  chain: {
+    eth1: IEth1ForBlockProduction;
+    executionEngine: IExecutionEngine;
+    config: IChainForkConfig;
+  },
   finalizedBlockHash: RootHex,
   state: CachedBeaconStateBellatrix,
   suggestedFeeRecipient: string
@@ -163,11 +171,20 @@ async function prepareExecutionPayload(
 
   const timestamp = computeTimeAtSlot(chain.config, state.slot, state.genesisTime);
   const prevRandao = getRandaoMix(state, state.epochCtx.epoch);
-  const payloadId = await chain.executionEngine.notifyForkchoiceUpdate(parentHash, finalizedBlockHash, {
-    timestamp,
-    prevRandao,
-    suggestedFeeRecipient,
-  });
+
+  const payloadId =
+    chain.executionEngine.payloadIdCache.get({
+      headBlockHash: toHex(parentHash),
+      finalizedBlockHash,
+      timestamp: numToQuantity(timestamp),
+      prevRandao: toHex(prevRandao),
+      suggestedFeeRecipient,
+    }) ??
+    (await chain.executionEngine.notifyForkchoiceUpdate(parentHash, finalizedBlockHash, {
+      timestamp,
+      prevRandao,
+      suggestedFeeRecipient,
+    }));
   if (!payloadId) throw new Error("InvalidPayloadId: Null");
   return payloadId;
 }

packages/lodestar/src/executionEngine/disabled.ts

@@ -1,6 +1,8 @@
-import {IExecutionEngine} from "./interface";
+import {IExecutionEngine, PayloadIdCache} from "./interface";
 
 export class ExecutionEngineDisabled implements IExecutionEngine {
+  readonly payloadIdCache = new PayloadIdCache();
+
   async notifyNewPayload(): Promise<never> {
     throw Error("Execution engine disabled");
   }

packages/lodestar/src/executionEngine/http.ts

@@ -23,6 +23,7 @@ import {
   PayloadAttributes,
   ApiPayloadAttributes,
 } from "./interface";
+import {PayloadIdCache} from "./payloadIdCache";
 
 export type ExecutionEngineHttpOpts = {
   urls: string[];
@@ -56,6 +57,7 @@ export const defaultExecutionEngineHttpOpts: ExecutionEngineHttpOpts = {
  * https://github.com/ethereum/execution-apis/blob/v1.0.0-alpha.1/src/engine/interop/specification.md
  */
 export class ExecutionEngineHttp implements IExecutionEngine {
+  readonly payloadIdCache = new PayloadIdCache();
   private readonly rpc: IJsonRpcHttpClient;
 
   constructor(opts: ExecutionEngineHttpOpts, signal: AbortSignal, rpc?: IJsonRpcHttpClient) {
@@ -220,8 +222,16 @@ export class ExecutionEngineHttp implements IExecutionEngine {
     switch (status) {
       case ExecutePayloadStatus.VALID:
         // if payloadAttributes are provided, a valid payloadId is expected
-        if (payloadAttributes && (!payloadId || payloadId === "0x")) {
-          throw Error(`Received invalid payloadId=${payloadId}`);
+        if (apiPayloadAttributes) {
+          if (!payloadId || payloadId === "0x") {
+            throw Error(`Received invalid payloadId=${payloadId}`);
+          }
+
+          this.payloadIdCache.add(
+            {headBlockHash: headBlockHashData, finalizedBlockHash, ...apiPayloadAttributes},
+            payloadId
+          );
+          void this.prunePayloadIdCache();
         }
         return payloadId !== "0x" ? payloadId : null;
 
@@ -271,6 +281,10 @@ export class ExecutionEngineHttp implements IExecutionEngine {
 
     return parseExecutionPayload(executionPayloadRpc);
   }
+
+  async prunePayloadIdCache(): Promise<void> {
+    this.payloadIdCache.prune();
+  }
 }
 
 /* eslint-disable @typescript-eslint/naming-convention */

packages/lodestar/src/executionEngine/interface.ts

@@ -1,10 +1,8 @@
 import {bellatrix, Root, RootHex} from "@chainsafe/lodestar-types";
 
-import {DATA, QUANTITY} from "../eth1/provider/utils";
-// An execution engine can produce a payload id anywhere the the uint64 range
-// Since we do no processing with this id, we have no need to deserialize it
-export type PayloadId = string;
+import {PayloadIdCache, PayloadId, ApiPayloadAttributes} from "./payloadIdCache";
 
+export {PayloadIdCache, PayloadId, ApiPayloadAttributes};
 export enum ExecutePayloadStatus {
   /** given payload is valid */
   VALID = "VALID",
@@ -57,22 +55,14 @@ export type PayloadAttributes = {
   suggestedFeeRecipient: string;
 };
 
-export type ApiPayloadAttributes = {
-  /** QUANTITY, 64 Bits - value for the timestamp field of the new payload */
-  timestamp: QUANTITY;
-  /** DATA, 32 Bytes - value for the prevRandao field of the new payload */
-  prevRandao: DATA;
-  /** DATA, 20 Bytes - suggested value for the coinbase field of the new payload */
-  suggestedFeeRecipient: DATA;
-};
-
 /**
  * Execution engine represents an abstract protocol to interact with execution clients. Potential transports include:
  * - JSON RPC over network
  * - IPC
  * - Integrated code into the same binary
  */
 export interface IExecutionEngine {
+  payloadIdCache: PayloadIdCache;
   /**
    * A state transition function which applies changes to the self.execution_state.
    * Returns ``True`` iff ``execution_payload`` is valid with respect to ``self.execution_state``.

packages/lodestar/src/executionEngine/mock.ts

@@ -10,6 +10,7 @@ import {
   IExecutionEngine,
   PayloadId,
   PayloadAttributes,
+  PayloadIdCache,
 } from "./interface";
 
 const INTEROP_GAS_LIMIT = 30e6;
@@ -25,6 +26,7 @@ export class ExecutionEngineMock implements IExecutionEngine {
   // Public state to check if notifyForkchoiceUpdate() is called properly
   headBlockRoot = ZERO_HASH_HEX;
   finalizedBlockRoot = ZERO_HASH_HEX;
+  readonly payloadIdCache = new PayloadIdCache();
 
   private knownBlocks = new Map<RootHex, bellatrix.ExecutionPayload>();
   private preparingPayloads = new Map<number, bellatrix.ExecutionPayload>();

packages/lodestar/src/executionEngine/payloadIdCache.ts

@@ -0,0 +1,46 @@
+import {pruneSetToMax} from "../util/map";
+import {IMetrics} from "../metrics";
+import {SLOTS_PER_EPOCH} from "@chainsafe/lodestar-params";
+import {DATA, QUANTITY} from "../eth1/provider/utils";
+
+// Idealy this only need to be set to the max head reorgs number
+const MAX_PAYLOAD_IDS = SLOTS_PER_EPOCH;
+
+// An execution engine can produce a payload id anywhere the the uint64 range
+// Since we do no processing with this id, we have no need to deserialize it
+export type PayloadId = string;
+
+export type ApiPayloadAttributes = {
+  /** QUANTITY, 64 Bits - value for the timestamp field of the new payload */
+  timestamp: QUANTITY;
+  /** DATA, 32 Bytes - value for the prevRandao field of the new payload */
+  prevRandao: DATA;
+  /** DATA, 20 Bytes - suggested value for the coinbase field of the new payload */
+  suggestedFeeRecipient: DATA;
+};
+
+type FcuAttributes = {headBlockHash: DATA; finalizedBlockHash: DATA} & ApiPayloadAttributes;
+
+export class PayloadIdCache {
+  private readonly payloadIdByFcuAttributes = new Map<string, PayloadId>();
+  constructor(private readonly metrics?: IMetrics | null) {}
+
+  getKey({headBlockHash, finalizedBlockHash, timestamp, prevRandao, suggestedFeeRecipient}: FcuAttributes): string {
+    return `${headBlockHash}-${finalizedBlockHash}-${timestamp}-${prevRandao}-${suggestedFeeRecipient}`;
+  }
+
+  add(fcuAttributes: FcuAttributes, payloadId: PayloadId): void {
+    const key = this.getKey(fcuAttributes);
+    this.payloadIdByFcuAttributes.set(key, payloadId);
+  }
+
+  prune(): void {
+    // This is not so optimized function, but could maintain a 2d array may be?
+    pruneSetToMax(this.payloadIdByFcuAttributes, MAX_PAYLOAD_IDS);
+  }
+
+  get(fcuAttributes: FcuAttributes): PayloadId | undefined {
+    const key = this.getKey(fcuAttributes);
+    return this.payloadIdByFcuAttributes.get(key);
+  }
+}