feat: gloas range sync (rebased onto epbs-devnet-0)

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

@@ -224,7 +224,7 @@ export function getBeaconBlockApi({
           }
 
           try {
-            await verifyBlocksInEpoch.call(chain as BeaconChain, parentBlock, [blockForImport], {
+            await verifyBlocksInEpoch.call(chain as BeaconChain, parentBlock, [blockForImport], null, {
               ...opts,
               verifyOnly: true,
               skipVerifyBlockSignatures: true,

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

@@ -113,7 +113,7 @@ export function getLodestarApi({
       return {
         // biome-ignore lint/complexity/useLiteralKeys: The `blockProcessor` is a protected attribute
         data: (chain as BeaconChain)["blockProcessor"].jobQueue.getItems().map((item) => {
-          const [blockInputs, opts] = item.args;
+          const [blockInputs, _envelopes, opts] = item.args;
           return {
             blockSlots: blockInputs.map((blockInput) => blockInput.slot),
             jobOpts: opts,

packages/beacon-node/src/chain/archiveStore/utils/archiveBlocks.ts

@@ -1,10 +1,10 @@
 import path from "node:path";
 import {ChainForkConfig} from "@lodestar/config";
 import {KeyValue} from "@lodestar/db";
-import {IForkChoice} from "@lodestar/fork-choice";
+import {CheckpointWithPayload, IForkChoice, PayloadStatus} from "@lodestar/fork-choice";
 import {ForkSeq, SLOTS_PER_EPOCH} from "@lodestar/params";
 import {computeEpochAtSlot, computeStartSlotAtEpoch} from "@lodestar/state-transition";
-import {Epoch, RootHex, Slot} from "@lodestar/types";
+import {Epoch, Slot} from "@lodestar/types";
 import {Logger, fromAsync, fromHex, prettyPrintIndices, toRootHex} from "@lodestar/utils";
 import {IBeaconDb} from "../../../db/index.js";
 import {BlockArchiveBatchPutBinaryItem} from "../../../db/repositories/index.js";
@@ -19,7 +19,6 @@ const BLOCK_BATCH_SIZE = 256;
 const BLOB_SIDECAR_BATCH_SIZE = 32;
 
 type BlockRootSlot = {slot: Slot; root: Uint8Array};
-type CheckpointHex = {epoch: Epoch; rootHex: RootHex};
 
 /**
  * Persist orphaned block to disk
@@ -53,7 +52,7 @@ export async function archiveBlocks(
   forkChoice: IForkChoice,
   lightclientServer: LightClientServer | undefined,
   logger: Logger,
-  finalizedCheckpoint: CheckpointHex,
+  finalizedCheckpoint: CheckpointWithPayload,
   currentEpoch: Epoch,
   archiveDataEpochs?: number,
   persistOrphanedBlocks?: boolean,
@@ -62,18 +61,36 @@ export async function archiveBlocks(
   // Use fork choice to determine the blocks to archive and delete
   // getAllAncestorBlocks response includes the finalized block, so it's also moved to the cold db
   const {ancestors: finalizedCanonicalBlocks, nonAncestors: finalizedNonCanonicalBlocks} =
-    forkChoice.getAllAncestorAndNonAncestorBlocks(finalizedCheckpoint.rootHex);
+    forkChoice.getAllAncestorAndNonAncestorBlocks(finalizedCheckpoint.rootHex, finalizedCheckpoint.payloadStatus);
 
   // NOTE: The finalized block will be exactly the first block of `epoch` or previous
   const finalizedPostDeneb = finalizedCheckpoint.epoch >= config.DENEB_FORK_EPOCH;
   const finalizedPostFulu = finalizedCheckpoint.epoch >= config.FULU_FORK_EPOCH;
+  const finalizedPostGloas = finalizedCheckpoint.epoch >= config.GLOAS_FORK_EPOCH;
+
+  const finalizedCanonicalBlockRoots: BlockRootSlot[] = [];
+  const finalizedCanonicalEnvelopeBlockRoots: BlockRootSlot[] = [];
+  for (const block of finalizedCanonicalBlocks) {
+    const rootAndSlot = {slot: block.slot, root: fromHex(block.blockRoot)};
+    const isPostGloasBlock = config.getForkSeq(block.slot) >= ForkSeq.gloas;
+
+    if (isPostGloasBlock) {
+      finalizedCanonicalBlockRoots.push(rootAndSlot);
+      if (block.payloadStatus === PayloadStatus.FULL) {
+        finalizedCanonicalEnvelopeBlockRoots.push(rootAndSlot);
+      }
+    } else {
+      finalizedCanonicalBlockRoots.push(rootAndSlot);
+    }
+  }
 
-  const finalizedCanonicalBlockRoots: BlockRootSlot[] = finalizedCanonicalBlocks.map((block) => ({
-    slot: block.slot,
-    root: fromHex(block.blockRoot),
-  }));
-
-  const logCtx = {currentEpoch, finalizedEpoch: finalizedCheckpoint.epoch, finalizedRoot: finalizedCheckpoint.rootHex};
+  const envelopeCount = finalizedCanonicalEnvelopeBlockRoots.length;
+  const logCtx = {
+    currentEpoch,
+    finalizedEpoch: finalizedCheckpoint.epoch,
+    finalizedRoot: finalizedCheckpoint.rootHex,
+    envelopeCount,
+  };
 
   if (finalizedCanonicalBlockRoots.length > 0) {
     await migrateBlocksFromHotToColdDb(db, finalizedCanonicalBlockRoots);
@@ -104,6 +121,14 @@ export async function archiveBlocks(
       );
       logger.verbose("Migrated dataColumnSidecars from hot DB to cold DB", {...logCtx, migratedEntries});
     }
+
+    if (finalizedPostGloas && finalizedCanonicalEnvelopeBlockRoots.length > 0) {
+      const migratedEntries = await migrateExecutionPayloadEnvelopesFromHotToColdDb(
+        db,
+        finalizedCanonicalEnvelopeBlockRoots
+      );
+      logger.verbose("Migrated executionPayloadEnvelopes from hot DB to cold DB", {...logCtx, migratedEntries});
+    }
   }
 
   // deleteNonCanonicalBlocks
@@ -373,6 +398,39 @@ async function migrateDataColumnSidecarsFromHotToColdDb(
   return migratedWrappedDataColumns;
 }
 
+async function migrateExecutionPayloadEnvelopesFromHotToColdDb(
+  db: IBeaconDb,
+  blocks: BlockRootSlot[]
+): Promise<number> {
+  let migratedEnvelopes = 0;
+  for (let i = 0; i < blocks.length; i += BLOCK_BATCH_SIZE) {
+    const toIdx = Math.min(i + BLOCK_BATCH_SIZE, blocks.length);
+    const canonicalBlocks = blocks.slice(i, toIdx);
+
+    if (canonicalBlocks.length === 0) break;
+
+    const canonicalEnvelopeEntries: KeyValue<Slot, Uint8Array>[] = await Promise.all(
+      canonicalBlocks.map(async (block) => {
+        const envelopeBytes = await db.executionPayloadEnvelope.getBinary(block.root);
+        if (!envelopeBytes) {
+          throw Error(`No executionPayloadEnvelope found for slot ${block.slot} root ${toRootHex(block.root)}`);
+        }
+
+        return {key: block.slot, value: envelopeBytes};
+      })
+    );
+
+    await Promise.all([
+      db.executionPayloadEnvelopeArchive.batchPutBinary(canonicalEnvelopeEntries),
+      db.executionPayloadEnvelope.batchDelete(canonicalBlocks.map((block) => block.root)),
+    ]);
+
+    migratedEnvelopes += canonicalEnvelopeEntries.length;
+  }
+
+  return migratedEnvelopes;
+}
+
 /**
  * ```
  * class SignedBeaconBlock(Container):

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

@@ -69,8 +69,15 @@ export async function importBlock(
   fullyVerifiedBlock: FullyVerifiedBlock,
   opts: ImportBlockOpts
 ): Promise<void> {
-  const {blockInput, postState, parentBlockSlot, executionStatus, dataAvailabilityStatus, indexedAttestations} =
-    fullyVerifiedBlock;
+  const {
+    blockInput,
+    postState,
+    postEnvelopeState,
+    parentBlockSlot,
+    executionStatus,
+    dataAvailabilityStatus,
+    indexedAttestations,
+  } = fullyVerifiedBlock;
   const block = blockInput.getBlock();
   const source = blockInput.getBlockSource();
   const {slot: blockSlot} = block.message;
@@ -122,9 +129,29 @@ export async function importBlock(
   const payloadPresent = !isGloasBlock;
   // processState manages both block state and payload state variants together for memory/disk management
   this.regen.processState(blockRootHex, postState);
+  this.logger.verbose("Added block to forkchoice and block state cache", {
+    slot: blockSlot,
+    root: blockRootHex,
+    stateRoot: toRootHex(postState.hashTreeRoot()),
+  });
+
+  if (postEnvelopeState !== null) {
+    this.regen.processPayloadState(postEnvelopeState);
+    this.forkChoice.onExecutionPayload(
+      blockRootHex,
+      toRootHex(postEnvelopeState.latestBlockHash),
+      // TODO GLOAS: this is not right but we don't need to track it as part of consensus spec, lighthouse also does not track it
+      0,
+      toRootHex(postEnvelopeState.hashTreeRoot())
+    );
+    this.logger.verbose("Added envelope state to block state cache", {
+      slot: blockSlot,
+      root: blockRootHex,
+      stateRoot: toRootHex(postEnvelopeState.hashTreeRoot()),
+    });
+  }
 
   this.metrics?.importBlock.bySource.inc({source: source.source});
-  this.logger.verbose("Added block to forkchoice and state cache", {slot: blockSlot, root: blockRootHex});
 
   // Post-Gloas: immediately import pending envelope for this block if available.
   // This makes the block FULL right away, so child blocks won't need to wait

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

@@ -1,4 +1,4 @@
-import {SignedBeaconBlock} from "@lodestar/types";
+import {SignedBeaconBlock, Slot, gloas} from "@lodestar/types";
 import {isErrorAborted, toRootHex} from "@lodestar/utils";
 import {Metrics} from "../../metrics/metrics.js";
 import {nextEventLoop} from "../../util/eventLoop.js";
@@ -22,20 +22,30 @@ const QUEUE_MAX_LENGTH = 256;
  * BlockProcessor processes block jobs in a queued fashion, one after the other.
  */
 export class BlockProcessor {
-  readonly jobQueue: JobItemQueue<[IBlockInput[], ImportBlockOpts], void>;
+  readonly jobQueue: JobItemQueue<
+    [IBlockInput[], Map<Slot, gloas.SignedExecutionPayloadEnvelope> | null, ImportBlockOpts],
+    void
+  >;
 
   constructor(chain: BeaconChain, metrics: Metrics | null, opts: BlockProcessOpts, signal: AbortSignal) {
-    this.jobQueue = new JobItemQueue<[IBlockInput[], ImportBlockOpts], void>(
-      (job, importOpts) => {
-        return processBlocks.call(chain, job, {...opts, ...importOpts});
+    this.jobQueue = new JobItemQueue<
+      [IBlockInput[], Map<Slot, gloas.SignedExecutionPayloadEnvelope> | null, ImportBlockOpts],
+      void
+    >(
+      (job, envelopes, importOpts) => {
+        return processBlocks.call(chain, job, envelopes, {...opts, ...importOpts});
       },
       {maxLength: QUEUE_MAX_LENGTH, noYieldIfOneItem: true, signal},
       metrics?.blockProcessorQueue ?? undefined
     );
   }
 
-  async processBlocksJob(job: IBlockInput[], opts: ImportBlockOpts = {}): Promise<void> {
-    await this.jobQueue.push(job, opts);
+  async processBlocksJob(
+    job: IBlockInput[],
+    envelopes: Map<Slot, gloas.SignedExecutionPayloadEnvelope> | null = null,
+    opts: ImportBlockOpts = {}
+  ): Promise<void> {
+    await this.jobQueue.push(job, envelopes, opts);
   }
 }
 
@@ -52,6 +62,7 @@ export class BlockProcessor {
 export async function processBlocks(
   this: BeaconChain,
   blocks: IBlockInput[],
+  envelopes: Map<Slot, gloas.SignedExecutionPayloadEnvelope> | null,
   opts: BlockProcessOpts & ImportBlockOpts
 ): Promise<void> {
   if (blocks.length === 0) {
@@ -63,7 +74,7 @@ export async function processBlocks(
   }
 
   try {
-    const {relevantBlocks, parentSlots, parentBlock} = verifyBlocksSanityChecks(this, blocks, opts);
+    const {relevantBlocks, parentSlots, parentBlock} = verifyBlocksSanityChecks(this, blocks, opts, envelopes);
 
     // No relevant blocks, skip verifyBlocksInEpoch()
     if (relevantBlocks.length === 0 || parentBlock === null) {
@@ -73,8 +84,14 @@ export async function processBlocks(
 
     // Fully verify a block to be imported immediately after. Does not produce any side-effects besides adding intermediate
     // states in the state cache through regen.
-    const {postStates, dataAvailabilityStatuses, proposerBalanceDeltas, segmentExecStatus, indexedAttestationsByBlock} =
-      await verifyBlocksInEpoch.call(this, parentBlock, relevantBlocks, opts);
+    const {
+      postStates,
+      postEnvelopeStates,
+      dataAvailabilityStatuses,
+      proposerBalanceDeltas,
+      segmentExecStatus,
+      indexedAttestationsByBlock,
+    } = await verifyBlocksInEpoch.call(this, parentBlock, relevantBlocks, envelopes, opts);
 
     // If segmentExecStatus has lvhForkchoice then, the entire segment should be invalid
     // and we need to further propagate
@@ -90,6 +107,7 @@ export async function processBlocks(
       (block, i): FullyVerifiedBlock => ({
         blockInput: block,
         postState: postStates[i],
+        postEnvelopeState: postEnvelopeStates.get(block.slot) ?? null,
         parentBlockSlot: parentSlots[i],
         executionStatus: executionStatuses[i],
         // start supporting optimistic syncing/processing

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

@@ -1,7 +1,12 @@
 import type {ChainForkConfig} from "@lodestar/config";
 import {MaybeValidExecutionStatus} from "@lodestar/fork-choice";
 import {ForkSeq} from "@lodestar/params";
-import {CachedBeaconStateAllForks, DataAvailabilityStatus, computeEpochAtSlot} from "@lodestar/state-transition";
+import {
+  CachedBeaconStateAllForks,
+  CachedBeaconStateGloas,
+  DataAvailabilityStatus,
+  computeEpochAtSlot,
+} from "@lodestar/state-transition";
 import type {IndexedAttestation, Slot, fulu} from "@lodestar/types";
 import {IBlockInput} from "./blockInput/types.js";
 
@@ -86,6 +91,7 @@ export type ImportBlockOpts = {
 export type FullyVerifiedBlock = {
   blockInput: IBlockInput;
   postState: CachedBeaconStateAllForks;
+  postEnvelopeState: CachedBeaconStateGloas | null;
   parentBlockSlot: Slot;
   proposerBalanceDelta: number;
   /**

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

@@ -2,11 +2,12 @@ import {ExecutionStatus, PayloadStatus, ProtoBlock} from "@lodestar/fork-choice"
 import {ForkName, isForkPostFulu} from "@lodestar/params";
 import {
   CachedBeaconStateAllForks,
+  CachedBeaconStateGloas,
   DataAvailabilityStatus,
   computeEpochAtSlot,
   isStateValidatorsNodesPopulated,
 } from "@lodestar/state-transition";
-import {IndexedAttestation, deneb, isGloasBeaconBlock} from "@lodestar/types";
+import {IndexedAttestation, Slot, deneb, gloas, isGloasBeaconBlock} from "@lodestar/types";
 import {sleep, toRootHex} from "@lodestar/utils";
 import type {BeaconChain} from "../chain.js";
 import {BlockError, BlockErrorCode} from "../errors/index.js";
@@ -38,9 +39,11 @@ export async function verifyBlocksInEpoch(
   this: BeaconChain,
   parentBlock: ProtoBlock,
   blockInputs: IBlockInput[],
+  envelopes: Map<Slot, gloas.SignedExecutionPayloadEnvelope> | null,
   opts: BlockProcessOpts & ImportBlockOpts
 ): Promise<{
   postStates: CachedBeaconStateAllForks[];
+  postEnvelopeStates: Map<Slot, CachedBeaconStateGloas | null>;
   proposerBalanceDeltas: number[];
   segmentExecStatus: SegmentExecStatus;
   dataAvailabilityStatuses: DataAvailabilityStatus[];
@@ -151,7 +154,15 @@ export async function verifyBlocksInEpoch(
     // Start execution payload verification first (async request to execution client)
     const verifyExecutionPayloadsPromise =
       opts.skipVerifyExecutionPayload !== true
-        ? verifyBlocksExecutionPayload(this, parentBlock, blockInputs, preState0, abortController.signal, opts)
+        ? verifyBlocksExecutionPayload(
+            this,
+            parentBlock,
+            blockInputs,
+            envelopes,
+            preState0,
+            abortController.signal,
+            opts
+          )
         : Promise.resolve({
             execAborted: null,
             executionStatuses: blocks.map((_blk) => ExecutionStatus.Syncing),
@@ -171,7 +182,7 @@ export async function verifyBlocksInEpoch(
     const [
       segmentExecStatus,
       {dataAvailabilityStatuses, availableTime},
-      {postStates, proposerBalanceDeltas, verifyStateTime},
+      {postStates, postEnvelopeStates, proposerBalanceDeltas, verifyStateTime},
       {verifySignaturesTime},
     ] = await Promise.all([
       verifyExecutionPayloadsPromise,
@@ -184,6 +195,7 @@ export async function verifyBlocksInEpoch(
       verifyBlocksStateTransitionOnly(
         preState0,
         blockInputs,
+        envelopes,
         // hack availability for state transition eval as availability is separately determined
         blocks.map(() => DataAvailabilityStatus.Available),
         this.logger,
@@ -202,6 +214,7 @@ export async function verifyBlocksInEpoch(
             this.metrics,
             preState0,
             blocks,
+            envelopes,
             indexedAttestationsByBlock,
             opts
           )
@@ -286,7 +299,14 @@ export async function verifyBlocksInEpoch(
       );
     }
 
-    return {postStates, dataAvailabilityStatuses, proposerBalanceDeltas, segmentExecStatus, indexedAttestationsByBlock};
+    return {
+      postStates,
+      postEnvelopeStates,
+      dataAvailabilityStatuses,
+      proposerBalanceDeltas,
+      segmentExecStatus,
+      indexedAttestationsByBlock,
+    };
   } finally {
     abortController.abort();
   }