EIP-4844: Update produce block flow

packages/beacon-node/src/api/impl/beacon/blocks/index.ts

@@ -1,11 +1,12 @@
 import {routes} from "@lodestar/api";
-
 import {computeTimeAtSlot} from "@lodestar/state-transition";
-import {SLOTS_PER_HISTORICAL_ROOT} from "@lodestar/params";
+import {ForkSeq, SLOTS_PER_HISTORICAL_ROOT} from "@lodestar/params";
 import {sleep} from "@lodestar/utils";
+import {eip4844} from "@lodestar/types";
 import {fromHexString, toHexString} from "@chainsafe/ssz";
-import {BlockError, BlockErrorCode} from "../../../../chain/errors/index.js";
 import {getBlockInput} from "../../../../chain/blocks/types.js";
+import {promiseAllMaybeAsync} from "../../../../util/promises.js";
+import {BlockError, BlockErrorCode} from "../../../../chain/errors/index.js";
 import {OpSource} from "../../../../metrics/validatorMonitor.js";
 import {NetworkEvent} from "../../../../network/index.js";
 import {ApiModules, IS_OPTIMISTIC_TEMP} from "../../types.js";
@@ -187,20 +188,28 @@ export function getBeaconBlockApi({
 
       metrics?.registerBeaconBlock(OpSource.api, seenTimestampSec, signedBlock.message);
 
-      // TODO EIP-4844: Will throw an error for blocks post EIP-4844
-      const blockInput = getBlockInput.preEIP4844(config, signedBlock);
-
-      await Promise.all([
+      // TODO EIP-4844: Open question if broadcast to both block topic + block_and_blobs topic
+      const blockForImport =
+        config.getForkSeq(signedBlock.message.slot) >= ForkSeq.eip4844
+          ? getBlockInput.postEIP4844(
+              config,
+              signedBlock,
+              chain.getBlobsSidecar(signedBlock.message as eip4844.BeaconBlock)
+            )
+          : getBlockInput.preEIP4844(config, signedBlock);
+
+      await promiseAllMaybeAsync([
         // Send the block, regardless of whether or not it is valid. The API
         // specification is very clear that this is the desired behaviour.
-        network.gossip.publishBeaconBlock(signedBlock),
-
-        chain.processBlock(blockInput).catch((e) => {
-          if (e instanceof BlockError && e.type.code === BlockErrorCode.PARENT_UNKNOWN) {
-            network.events.emit(NetworkEvent.unknownBlockParent, blockInput, network.peerId.toString());
-          }
-          throw e;
-        }),
+        () => network.publishBeaconBlockMaybeBlobs(blockForImport),
+
+        () =>
+          chain.processBlock(blockForImport).catch((e) => {
+            if (e instanceof BlockError && e.type.code === BlockErrorCode.PARENT_UNKNOWN) {
+              network.events.emit(NetworkEvent.unknownBlockParent, blockForImport, network.peerId.toString());
+            }
+            throw e;
+          }),
       ]);
     },
   };

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

@@ -1,4 +1,5 @@
 import path from "node:path";
+import {computeAggregateKzgProof} from "c-kzg";
 import {
   BeaconStateAllForks,
   CachedBeaconStateAllForks,
@@ -12,12 +13,12 @@ import {
   PubkeyIndexMap,
 } from "@lodestar/state-transition";
 import {IBeaconConfig} from "@lodestar/config";
-import {allForks, UintNum64, Root, phase0, Slot, RootHex, Epoch, ValidatorIndex} from "@lodestar/types";
+import {allForks, UintNum64, Root, phase0, Slot, RootHex, Epoch, ValidatorIndex, eip4844} from "@lodestar/types";
 import {CheckpointWithHex, ExecutionStatus, IForkChoice, ProtoBlock} from "@lodestar/fork-choice";
 import {ProcessShutdownCallback} from "@lodestar/validator";
-import {ILogger, toHex} from "@lodestar/utils";
+import {ILogger, pruneSetToMax, toHex} from "@lodestar/utils";
 import {CompositeTypeAny, fromHexString, TreeView, Type} from "@chainsafe/ssz";
-import {SLOTS_PER_EPOCH} from "@lodestar/params";
+import {ForkSeq, SLOTS_PER_EPOCH} from "@lodestar/params";
 
 import {GENESIS_EPOCH, ZERO_HASH} from "../constants/index.js";
 import {IBeaconDb} from "../db/index.js";
@@ -59,11 +60,18 @@ import {SeenAggregatedAttestations} from "./seenCache/seenAggregateAndProof.js";
 import {SeenBlockAttesters} from "./seenCache/seenBlockAttesters.js";
 import {BeaconProposerCache} from "./beaconProposerCache.js";
 import {CheckpointBalancesCache} from "./balancesCache.js";
-import {AssembledBlockType, BlockType} from "./produceBlock/index.js";
+import {AssembledBlockType, BlobsResultType, BlockType} from "./produceBlock/index.js";
 import {BlockAttributes, produceBlockBody} from "./produceBlock/produceBlockBody.js";
 import {computeNewStateRoot} from "./produceBlock/computeNewStateRoot.js";
 import {BlockInput} from "./blocks/types.js";
 
+/**
+ * Arbitrary constants, blobs should be consumed immediately in the same slot they are produced.
+ * A value of 1 would probably be sufficient. However it's sensible to allow some margin if the node overloads.
+ */
+const DEFAULT_MAX_CACHED_BLOBS_SIDECAR = 8;
+const MAX_RETAINED_SLOTS_CACHED_BLOBS_SIDECAR = 8;
+
 export class BeaconChain implements IBeaconChain {
   readonly genesisTime: UintNum64;
   readonly genesisValidatorsRoot: Root;
@@ -119,6 +127,10 @@ export class BeaconChain implements IBeaconChain {
   private successfulExchangeTransition = false;
   private readonly exchangeTransitionConfigurationEverySlots: number;
 
+  // TODO EIP-4844: Prune data structure every time period, for both old entries
+  /** Map keyed by executionPayload.blockHash of the block for those blobs */
+  private readonly producedBlobsSidecarCache = new Map<RootHex, eip4844.BlobsSidecar>();
+
   private readonly faultInspectionWindow: number;
   private readonly allowedFaults: number;
   private processShutdownCallback: ProcessShutdownCallback;
@@ -359,27 +371,65 @@ export class BeaconChain implements IBeaconChain {
     const proposerIndex = state.epochCtx.getBeaconProposer(slot);
     const proposerPubKey = state.epochCtx.index2pubkey[proposerIndex].toBytes();
 
+    const {body, blobs} = await produceBlockBody.call(this, blockType, state, {
+      randaoReveal,
+      graffiti,
+      slot,
+      parentSlot: slot - 1,
+      parentBlockRoot,
+      proposerIndex,
+      proposerPubKey,
+    });
+
     const block = {
       slot,
       proposerIndex,
       parentRoot: parentBlockRoot,
       stateRoot: ZERO_HASH,
-      body: await produceBlockBody.call(this, blockType, state, {
-        randaoReveal,
-        graffiti,
-        slot,
-        parentSlot: slot - 1,
-        parentBlockRoot,
-        proposerIndex,
-        proposerPubKey,
-      }),
+      body,
     } as AssembledBlockType<T>;
 
     block.stateRoot = computeNewStateRoot(this.metrics, state, block);
 
+    // Cache for latter broadcasting
+    if (blobs.type === BlobsResultType.produced) {
+      // TODO EIP-4844: Prune data structure for max entries
+      this.producedBlobsSidecarCache.set(blobs.blockHash, {
+        // TODO EIP-4844: Optimize, hashing the full block is not free.
+        beaconBlockRoot: this.config.getForkTypes(block.slot).BeaconBlock.hashTreeRoot(block),
+        beaconBlockSlot: block.slot,
+        blobs: blobs.blobs,
+        kzgAggregatedProof: computeAggregateKzgProof(blobs.blobs),
+      });
+      pruneSetToMax(
+        this.producedBlobsSidecarCache,
+        this.opts.maxCachedBlobsSidecar ?? DEFAULT_MAX_CACHED_BLOBS_SIDECAR
+      );
+    }
+
     return block;
   }
 
+  /**
+   * https://github.com/ethereum/consensus-specs/blob/dev/specs/eip4844/validator.md#sidecar
+   * def get_blobs_sidecar(block: BeaconBlock, blobs: Sequence[Blob]) -> BlobsSidecar:
+   *   return BlobsSidecar(
+   *       beacon_block_root=hash_tree_root(block),
+   *       beacon_block_slot=block.slot,
+   *       blobs=blobs,
+   *       kzg_aggregated_proof=compute_proof_from_blobs(blobs),
+   *   )
+   */
+  getBlobsSidecar(beaconBlock: eip4844.BeaconBlock): eip4844.BlobsSidecar {
+    const blockHash = toHex(beaconBlock.body.executionPayload.blockHash);
+    const blobsSidecar = this.producedBlobsSidecarCache.get(blockHash);
+    if (!blobsSidecar) {
+      throw Error(`No blobsSidecar for executionPayload.blockHash ${blockHash}`);
+    }
+
+    return blobsSidecar;
+  }
+
   async processBlock(block: BlockInput, opts?: ImportBlockOpts): Promise<void> {
     return await this.blockProcessor.processBlocksJob([block], opts);
   }
@@ -591,6 +641,15 @@ export class BeaconChain implements IBeaconChain {
         this.logger.error("Error on exchangeTransitionConfiguration", {}, e as Error);
       });
     }
+
+    // Prune old blobsSidecar for block production, those are only useful on their slot
+    if (this.config.getForkSeq(slot) >= ForkSeq.eip4844 && this.producedBlobsSidecarCache.size > 0) {
+      for (const [key, blobsSidecar] of this.producedBlobsSidecarCache) {
+        if (slot > blobsSidecar.beaconBlockSlot + MAX_RETAINED_SLOTS_CACHED_BLOBS_SIDECAR) {
+          this.producedBlobsSidecarCache.delete(key);
+        }
+      }
+    }
   }
 
   private onClockEpoch(epoch: Epoch): void {

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

@@ -1,4 +1,4 @@
-import {allForks, UintNum64, Root, phase0, Slot, RootHex, Epoch, ValidatorIndex} from "@lodestar/types";
+import {allForks, UintNum64, Root, phase0, Slot, RootHex, Epoch, ValidatorIndex, eip4844} from "@lodestar/types";
 import {CachedBeaconStateAllForks} from "@lodestar/state-transition";
 import {IBeaconConfig} from "@lodestar/config";
 import {CompositeTypeAny, TreeView, Type} from "@chainsafe/ssz";
@@ -110,6 +110,8 @@ export interface IBeaconChain {
    */
   getCanonicalBlockAtSlot(slot: Slot): Promise<allForks.SignedBeaconBlock | null>;
 
+  getBlobsSidecar(beaconBlock: eip4844.BeaconBlock): eip4844.BlobsSidecar;
+
   produceBlock(blockAttributes: BlockAttributes): Promise<allForks.BeaconBlock>;
   produceBlindedBlock(blockAttributes: BlockAttributes): Promise<allForks.BlindedBeaconBlock>;
 

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

@@ -20,6 +20,10 @@ export type IChainOptions = BlockProcessOpts &
     faultInspectionWindow?: number;
     /** Number of missed slots allowed in the faultInspectionWindow for builder circuit*/
     allowedFaults?: number;
+    /** Ensure blobs returned by the execution engine are valid */
+    sanityCheckExecutionEngineBlobs?: boolean;
+    /** Max number of produced blobs by local validators to cache */
+    maxCachedBlobsSidecar?: number;
   };
 
 export type BlockProcessOpts = {

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

@@ -11,6 +11,7 @@ import {
   BLSPubkey,
   BLSSignature,
   capella,
+  eip4844,
 } from "@lodestar/types";
 import {
   CachedBeaconStateAllForks,
@@ -39,6 +40,7 @@ import {
 import {ZERO_HASH, ZERO_HASH_HEX} from "../../constants/index.js";
 import {IEth1ForBlockProduction} from "../../eth1/index.js";
 import {numToQuantity} from "../../eth1/provider/utils.js";
+import {validateBlobsAndKzgCommitments} from "./validateBlobsAndKzgCommitments.js";
 
 // Time to provide the EL to generate a payload from new payload id
 const PAYLOAD_GENERATION_TIME_MS = 500;
@@ -65,6 +67,15 @@ export type AssembledBlockType<T extends BlockType> = T extends BlockType.Full
   ? allForks.BeaconBlock
   : allForks.BlindedBeaconBlock;
 
+export enum BlobsResultType {
+  preEIP4844,
+  produced,
+}
+
+export type BlobsResult =
+  | {type: BlobsResultType.preEIP4844}
+  | {type: BlobsResultType.produced; blobs: eip4844.Blobs; blockHash: RootHex};
+
 export async function produceBlockBody<T extends BlockType>(
   this: BeaconChain,
   blockType: T,
@@ -83,7 +94,10 @@ export async function produceBlockBody<T extends BlockType>(
     proposerIndex: ValidatorIndex;
     proposerPubKey: BLSPubkey;
   }
-): Promise<AssembledBodyType<T>> {
+): Promise<{body: AssembledBodyType<T>; blobs: BlobsResult}> {
+  // We assign this in an EIP-4844 branch below and return it
+  let blobs: {blobs: eip4844.Blobs; blockHash: RootHex} | null = null;
+
   // TODO:
   // Iterate through the naive aggregation pool and ensure all the attestations from there
   // are included in the operation pool.
@@ -153,7 +167,16 @@ export async function produceBlockBody<T extends BlockType>(
         currentState as CachedBeaconStateBellatrix,
         proposerPubKey
       );
-    } else {
+
+      // Capella and later forks have withdrawalRoot on their ExecutionPayloadHeader
+      // TODO Capella: Remove this. It will come from the execution client.
+      if (ForkSeq[fork] >= ForkSeq.capella) {
+        throw Error("Builder blinded blocks not supported after capella");
+      }
+    }
+
+    // blockType === BlockType.Full
+    else {
       // try catch payload fetch here, because there is still a recovery path possible if we
       // are pre-merge. We don't care the same for builder segment as the execution block
       // will takeover if the builder flow was activated and errors
@@ -188,6 +211,28 @@ export async function produceBlockBody<T extends BlockType>(
           if (payload.transactions.length === 0) {
             this.metrics?.blockPayload.emptyPayloads.inc({prepType});
           }
+
+          if (fork === ForkName.eip4844) {
+            // SPEC: https://github.com/ethereum/consensus-specs/blob/dev/specs/eip4844/validator.md#blob-kzg-commitments
+            // After retrieving the execution payload from the execution engine as specified in Bellatrix, use the
+            // payload_id to retrieve blobs and blob_kzg_commitments via get_blobs_and_kzg_commitments(payload_id)
+            // TODO EIP-4844: getBlobsBundle and getPayload must be either coupled or called in parallel to save time.
+            const blobsBundle = await this.executionEngine.getBlobsBundle(payloadId);
+
+            // Sanity check consistency between getPayload() and getBlobsBundle()
+            const blockHash = toHex(payload.blockHash);
+            if (blobsBundle.blockHash !== blockHash) {
+              throw Error(`blobsBundle incorrect blockHash ${blobsBundle.blockHash} != ${blockHash}`);
+            }
+
+            // Optionally sanity-check that the KZG commitments match the versioned hashes in the transactions
+            if (this.opts.sanityCheckExecutionEngineBlobs) {
+              validateBlobsAndKzgCommitments(payload, blobsBundle);
+            }
+
+            (blockBody as eip4844.BeaconBlockBody).blobKzgCommitments = blobsBundle.kzgs;
+            blobs = {blobs: blobsBundle.blobs, blockHash};
+          }
         }
       } catch (e) {
         this.metrics?.blockPayload.payloadFetchErrors.inc();
@@ -217,7 +262,20 @@ export async function produceBlockBody<T extends BlockType>(
     (blockBody as capella.BeaconBlockBody).blsToExecutionChanges = [];
   }
 
-  return blockBody as AssembledBodyType<T>;
+  // Type-safe for blobs variable. Translate 'null' value into 'preEIP4844' enum
+  // TODO: Not ideal, but better than just using null.
+  // TODO: Does not guarantee that preEIP4844 enum goes with a preEIP4844 block
+  let blobsResult: BlobsResult;
+  if (currentState.config.getForkSeq(blockSlot) >= ForkSeq.eip4844) {
+    if (!blobs) {
+      throw Error("Blobs are null post eip4844");
+    }
+    blobsResult = {type: BlobsResultType.produced, ...blobs};
+  } else {
+    blobsResult = {type: BlobsResultType.preEIP4844};
+  }
+
+  return {body: blockBody as AssembledBodyType<T>, blobs: blobsResult};
 }
 
 /**

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

@@ -0,0 +1,26 @@
+import {blobToKzgCommitment} from "c-kzg";
+import {verifyKzgCommitmentsAgainstTransactions} from "@lodestar/state-transition";
+import {allForks, eip4844} from "@lodestar/types";
+import {toHex} from "@lodestar/utils";
+import {BlobsBundle} from "../../execution/index.js";
+import {byteArrayEquals} from "../../util/bytes.js";
+
+/**
+ * Optionally sanity-check that the KZG commitments match the versioned hashes in the transactions
+ * https://github.com/ethereum/consensus-specs/blob/11a037fd9227e29ee809c9397b09f8cc3383a8c0/specs/eip4844/validator.md#blob-kzg-commitments
+ */
+export function validateBlobsAndKzgCommitments(payload: allForks.ExecutionPayload, blobsBundle: BlobsBundle): void {
+  verifyKzgCommitmentsAgainstTransactions(payload.transactions, blobsBundle.kzgs);
+
+  // Optionally sanity-check that the KZG commitments match the blobs (as produced by the execution engine)
+  if (blobsBundle.blobs.length !== blobsBundle.kzgs.length) {
+    throw Error(`Blobs bundle blobs len ${blobsBundle.blobs.length} != kzgs len ${blobsBundle.kzgs.length}`);
+  }
+
+  for (let i = 0; i < blobsBundle.blobs.length; i++) {
+    const kzg = blobToKzgCommitment(blobsBundle.blobs[i]) as eip4844.KZGCommitment;
+    if (!byteArrayEquals(kzg, blobsBundle.kzgs[i])) {
+      throw Error(`Wrong KZG[${i}] ${toHex(blobsBundle.kzgs[i])} expected ${toHex(kzg)}`);
+    }
+  }
+}

packages/beacon-node/src/network/interface.ts

@@ -32,6 +32,7 @@ export interface INetwork {
   getConnectedPeers(): PeerId[];
   hasSomeConnectedPeer(): boolean;
 
+  publishBeaconBlockMaybeBlobs(signedBlock: BlockInput): Promise<void>;
   beaconBlocksMaybeBlobsByRange(peerId: PeerId, request: phase0.BeaconBlocksByRangeRequest): Promise<BlockInput[]>;
   beaconBlocksMaybeBlobsByRoot(peerId: PeerId, request: phase0.BeaconBlocksByRootRequest): Promise<BlockInput[]>;