feat: free the blobs

.github/workflows/test-sim-merge.yml

@@ -17,8 +17,9 @@ env:
   NETHERMIND_IMAGE: nethermind/nethermind:1.14.3
   MERGEMOCK_IMAGE: g11tech/mergemock:latest
   GETH_WITHDRAWALS_IMAGE: g11tech/geth:withdrawalsfeb8
-  ETHEREUMJS_WITHDRAWALS_IMAGE: g11tech/ethereumjs:feb8
+  ETHEREUMJS_WITHDRAWALS_IMAGE: g11tech/ethereumjs:blobs-b6b63
   NETHERMIND_WITHDRAWALS_IMAGE: nethermindeth/nethermind:withdrawals_yolo
+  ETHEREUMJS_BLOBS_IMAGE: g11tech/ethereumjs:blobs-b6b63
 
 jobs:
   sim-merge-tests:
@@ -128,6 +129,18 @@ jobs:
       #     EL_BINARY_DIR: ${{ env.NETHERMIND_WITHDRAWALS_IMAGE }}
       #     EL_SCRIPT_DIR: netherminddocker
 
+      # Enable the blob sims when stable images
+      # - name: Pull ethereumjs blobs
+      #   run: docker pull $ETHEREUMJS_BLOBS_IMAGE
+
+      # - name: Test Lodestar <> ethereumjs blobs
+      #   run: yarn test:sim:blobs
+      #   working-directory: packages/beacon-node
+      #   env:
+      #     EL_BINARY_DIR: ${{ env.ETHEREUMJS_BLOBS_IMAGE }}
+      #     EL_SCRIPT_DIR: ethereumjsdocker
+      #     DEV_RUN: true
+
       - name: Upload debug log test files
         if: ${{ always() }}
         uses: actions/upload-artifact@v2

packages/beacon-node/package.json

@@ -85,6 +85,7 @@
     "test:sim:merge-interop": "mocha 'test/sim/merge-interop.test.ts'",
     "test:sim:mergemock": "mocha 'test/sim/mergemock.test.ts'",
     "test:sim:withdrawals": "mocha 'test/sim/withdrawal-interop.test.ts'",
+    "test:sim:blobs": "mocha 'test/sim/4844-interop.test.ts'",
     "download-spec-tests": "node --loader=ts-node/esm test/spec/downloadTests.ts",
     "check-spec-tests": "mocha test/spec/checkCoverage.ts",
     "test:spec-bls-general": "mocha --config .mocharc.spec.cjs 'test/spec/bls/**/*.test.ts' 'test/spec/general/**/*.test.ts'",

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

@@ -4,13 +4,7 @@ import {computeTimeAtSlot} from "@lodestar/state-transition";
 import {SLOTS_PER_HISTORICAL_ROOT} from "@lodestar/params";
 import {sleep, toHex} from "@lodestar/utils";
 import {allForks, deneb} from "@lodestar/types";
-import {
-  BlockSource,
-  getBlockInput,
-  ImportBlockOpts,
-  BlockInput,
-  blobSidecarsToBlobsSidecar,
-} from "../../../../chain/blocks/types.js";
+import {BlockSource, getBlockInput, ImportBlockOpts, BlockInput} from "../../../../chain/blocks/types.js";
 import {promiseAllMaybeAsync} from "../../../../util/promises.js";
 import {isOptimisticBlock} from "../../../../util/forkChoice.js";
 import {BlockError, BlockErrorCode} from "../../../../chain/errors/index.js";
@@ -52,13 +46,10 @@ export function getBeaconBlockApi({
         config,
         signedBlock,
         BlockSource.api,
-        // The blobsSidecar will be replaced in the followup PRs with just blobs
-        blobSidecarsToBlobsSidecar(
-          config,
-          signedBlock,
-          signedBlobs.map((sblob) => sblob.message)
-        ),
-        null
+        signedBlobs.map((sblob) => sblob.message),
+        // don't bundle any bytes for block and blobs
+        null,
+        signedBlobs.map(() => null)
       );
     } else {
       signedBlock = signedBlockOrContents;

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

@@ -1,10 +1,10 @@
+import {toHexString} from "@chainsafe/ssz";
 import {CachedBeaconStateAllForks, computeEpochAtSlot, DataAvailableStatus} from "@lodestar/state-transition";
 import {MaybeValidExecutionStatus} from "@lodestar/fork-choice";
-import {allForks, deneb, Slot} from "@lodestar/types";
+import {allForks, deneb, Slot, RootHex} from "@lodestar/types";
 import {ForkSeq, MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS} from "@lodestar/params";
 import {ChainForkConfig} from "@lodestar/config";
-
-import {ckzg} from "../../util/kzg.js";
+import {pruneSetToMax} from "@lodestar/utils";
 
 export enum BlockInputType {
   preDeneb = "preDeneb",
@@ -21,7 +21,7 @@ export enum BlockSource {
 
 export type BlockInput = {block: allForks.SignedBeaconBlock; source: BlockSource; blockBytes: Uint8Array | null} & (
   | {type: BlockInputType.preDeneb}
-  | {type: BlockInputType.postDeneb; blobs: deneb.BlobsSidecar}
+  | {type: BlockInputType.postDeneb; blobs: deneb.BlobSidecars; blobsBytes: (Uint8Array | null)[]}
 );
 
 export function blockRequiresBlobs(config: ChainForkConfig, blockSlot: Slot, clockSlot: Slot): boolean {
@@ -32,26 +32,126 @@ export function blockRequiresBlobs(config: ChainForkConfig, blockSlot: Slot, clo
   );
 }
 
-// TODO DENEB: a helper function to convert blobSidecars to blobsSidecar, to be cleanup on BlockInput
-// migration
-export function blobSidecarsToBlobsSidecar(
-  config: ChainForkConfig,
-  signedBlock: allForks.SignedBeaconBlock,
-  blobSidecars: deneb.BlobSidecars
-): deneb.BlobsSidecar {
-  const beaconBlockSlot = signedBlock.message.slot;
-  const beaconBlockRoot = config.getForkTypes(beaconBlockSlot).BeaconBlock.hashTreeRoot(signedBlock.message);
-  const blobs = blobSidecars.map(({blob}) => blob);
-  const blobsSidecar = {
-    beaconBlockRoot,
-    beaconBlockSlot,
-    blobs,
-    kzgAggregatedProof: ckzg.computeAggregateKzgProof(blobs),
-  };
-  return blobsSidecar;
+export enum GossipedInputType {
+  block = "block",
+  blob = "blob",
 }
+type GossipedBlockInput =
+  | {type: GossipedInputType.block; signedBlock: allForks.SignedBeaconBlock; blockBytes: Uint8Array | null}
+  | {type: GossipedInputType.blob; signedBlob: deneb.SignedBlobSidecar; blobBytes: Uint8Array | null};
+type BlockInputCacheType = {
+  block?: allForks.SignedBeaconBlock;
+  blockBytes?: Uint8Array | null;
+  blobs: Map<number, deneb.BlobSidecar>;
+  blobsBytes: Map<number, Uint8Array | null>;
+};
+
+const MAX_GOSSIPINPUT_CACHE = 5;
+// TODO deneb: export from types package
+// ssz.deneb.BlobSidecars.elementType.fixedSize;
+const BLOBSIDECAR_FIXED_SIZE = 131256;
 
 export const getBlockInput = {
+  blockInputCache: new Map<RootHex, BlockInputCacheType>(),
+
+  getGossipBlockInput(
+    config: ChainForkConfig,
+    gossipedInput: GossipedBlockInput
+  ):
+    | {blockInput: BlockInput; blockInputMeta: {pending: null; haveBlobs: number; expectedBlobs: number}}
+    | {blockInput: null; blockInputMeta: {pending: GossipedInputType.block; haveBlobs: number; expectedBlobs: null}}
+    | {blockInput: null; blockInputMeta: {pending: GossipedInputType.blob; haveBlobs: number; expectedBlobs: number}} {
+    let blockHex;
+    let blockCache;
+
+    if (gossipedInput.type === GossipedInputType.block) {
+      const {signedBlock, blockBytes} = gossipedInput;
+
+      blockHex = toHexString(
+        config.getForkTypes(signedBlock.message.slot).BeaconBlock.hashTreeRoot(signedBlock.message)
+      );
+      blockCache = this.blockInputCache.get(blockHex) ?? {
+        blobs: new Map<number, deneb.BlobSidecar>(),
+        blobsBytes: new Map<number, Uint8Array | null>(),
+      };
+
+      blockCache.block = signedBlock;
+      blockCache.blockBytes = blockBytes;
+    } else {
+      const {signedBlob, blobBytes} = gossipedInput;
+      blockHex = toHexString(signedBlob.message.blockRoot);
+      blockCache = this.blockInputCache.get(blockHex);
+
+      // If a new entry is going to be inserted, prune out old ones
+      if (blockCache === undefined) {
+        pruneSetToMax(this.blockInputCache, MAX_GOSSIPINPUT_CACHE);
+        blockCache = {blobs: new Map<number, deneb.BlobSidecar>(), blobsBytes: new Map<number, Uint8Array | null>()};
+      }
+
+      // TODO: freetheblobs check if its the same blob or a duplicate and throw/take actions
+      blockCache.blobs.set(signedBlob.message.index, signedBlob.message);
+      // easily splice out the unsigned message as blob is a fixed length type
+      blockCache.blobsBytes.set(signedBlob.message.index, blobBytes?.slice(0, BLOBSIDECAR_FIXED_SIZE) ?? null);
+    }
+
+    this.blockInputCache.set(blockHex, blockCache);
+    const {block: signedBlock, blockBytes} = blockCache;
+
+    if (signedBlock !== undefined) {
+      // block is available, check if all blobs have shown up
+      const {slot, body} = signedBlock.message;
+      const {blobKzgCommitments} = body as deneb.BeaconBlockBody;
+      const blockInfo = `blockHex=${blockHex}, slot=${slot}`;
+
+      if (blobKzgCommitments.length < blockCache.blobs.size) {
+        throw Error(
+          `Received more blobs=${blockCache.blobs.size} than commitments=${blobKzgCommitments.length} for ${blockInfo}`
+        );
+      }
+      if (blobKzgCommitments.length === blockCache.blobs.size) {
+        const blobSidecars = [];
+        const blobsBytes = [];
+
+        for (let index = 0; index < blobKzgCommitments.length; index++) {
+          const blobSidecar = blockCache.blobs.get(index);
+          if (blobSidecar === undefined) {
+            throw Error(`Missing blobSidecar at index=${index} for ${blockInfo}`);
+          }
+          blobSidecars.push(blobSidecar);
+          blobsBytes.push(blockCache.blobsBytes.get(index) ?? null);
+        }
+
+        return {
+          // TODO freetheblobs: collate and add serialized data for the postDeneb blockinput
+          blockInput: getBlockInput.postDeneb(
+            config,
+            signedBlock,
+            BlockSource.gossip,
+            blobSidecars,
+            blockBytes ?? null,
+            blobsBytes
+          ),
+          blockInputMeta: {pending: null, haveBlobs: blockCache.blobs.size, expectedBlobs: blobKzgCommitments.length},
+        };
+      } else {
+        return {
+          blockInput: null,
+          blockInputMeta: {
+            pending: GossipedInputType.blob,
+            haveBlobs: blockCache.blobs.size,
+            expectedBlobs: blobKzgCommitments.length,
+          },
+        };
+      }
+    } else {
+      // will need to wait for the block to showup
+      return {
+        blockInput: null,
+        blockInputMeta: {pending: GossipedInputType.block, haveBlobs: blockCache.blobs.size, expectedBlobs: null},
+      };
+    }
+  },
+
   preDeneb(
     config: ChainForkConfig,
     block: allForks.SignedBeaconBlock,
@@ -73,8 +173,9 @@ export const getBlockInput = {
     config: ChainForkConfig,
     block: allForks.SignedBeaconBlock,
     source: BlockSource,
-    blobs: deneb.BlobsSidecar,
-    blockBytes: Uint8Array | null
+    blobs: deneb.BlobSidecars,
+    blockBytes: Uint8Array | null,
+    blobsBytes: (Uint8Array | null)[]
   ): BlockInput {
     if (config.getForkSeq(block.message.slot) < ForkSeq.deneb) {
       throw Error(`Pre Deneb block slot ${block.message.slot}`);
@@ -85,6 +186,7 @@ export const getBlockInput = {
       source,
       blobs,
       blockBytes,
+      blobsBytes,
     };
   },
 };
@@ -127,7 +229,7 @@ export type ImportBlockOpts = {
    * Metadata: `true` if all the signatures including the proposer signature have been verified
    */
   validSignatures?: boolean;
-  /** Set to true if already run `validateBlobsSidecar()` sucessfully on the blobs */
+  /** Set to true if already run `validateBlobSidecars()` sucessfully on the blobs */
   validBlobSidecars?: boolean;
   /** Seen timestamp seconds */
   seenTimestampSec?: number;

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

@@ -5,8 +5,6 @@ import {Slot, deneb} from "@lodestar/types";
 import {toHexString} from "@lodestar/utils";
 import {IClock} from "../../util/clock.js";
 import {BlockError, BlockErrorCode} from "../errors/index.js";
-// TODO freetheblobs: disable the following exception once blockinput changes
-/* eslint-disable @typescript-eslint/no-unused-vars */
 import {validateBlobSidecars} from "../validation/blobSidecar.js";
 import {BlockInput, BlockInputType, ImportBlockOpts} from "./types.js";
 
@@ -137,8 +135,7 @@ function maybeValidateBlobs(
       const {blobKzgCommitments} = (block as deneb.SignedBeaconBlock).message.body;
       const beaconBlockRoot = config.getForkTypes(blockSlot).BeaconBlock.hashTreeRoot(block.message);
       // TODO Deneb: This function throws un-typed errors
-      // TODO freetheblobs: enable the following validation once blockinput is migrated
-      // validateBlobSidecars(blockSlot, beaconBlockRoot, blobKzgCommitments, blobs);
+      validateBlobSidecars(blockSlot, beaconBlockRoot, blobKzgCommitments, blobs);
 
       return DataAvailableStatus.available;
     }

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

@@ -1,4 +1,3 @@
-import {allForks, deneb} from "@lodestar/types";
 import {toHex} from "@lodestar/utils";
 import {BeaconChain} from "../chain.js";
 import {BlockInput, BlockInputType} from "./types.js";
@@ -31,13 +30,13 @@ export async function writeBlockInputToDb(this: BeaconChain, blocksInput: BlockI
     });
 
     if (type === BlockInputType.postDeneb) {
-      const {blobs} = blockInput;
+      const {blobs: blobSidecars} = blockInput;
       // NOTE: Old blobs are pruned on archive
-      fnPromises.push(this.db.blobsSidecar.add(blobs));
-      this.logger.debug("Persist blobsSidecar to hot DB", {
-        blobsLen: blobs.blobs.length,
-        slot: blobs.beaconBlockSlot,
-        root: toHex(blobs.beaconBlockRoot),
+      fnPromises.push(this.db.blobSidecars.add({blockRoot, slot: block.message.slot, blobSidecars}));
+      this.logger.debug("Persisted blobSidecars to hot DB", {
+        blobsLen: blobSidecars.length,
+        slot: block.message.slot,
+        root: blockRootHex,
       });
     }
   }
@@ -49,27 +48,26 @@ export async function writeBlockInputToDb(this: BeaconChain, blocksInput: BlockI
  * Prunes eagerly persisted block inputs only if not known to the fork-choice
  */
 export async function removeEagerlyPersistedBlockInputs(this: BeaconChain, blockInputs: BlockInput[]): Promise<void> {
-  const blockToRemove: allForks.SignedBeaconBlock[] = [];
-  const blobsToRemove: deneb.BlobsSidecar[] = [];
+  const blockToRemove = [];
+  const blobsToRemove = [];
 
   for (const blockInput of blockInputs) {
     const {block, type} = blockInput;
-    const blockRoot = toHex(this.config.getForkTypes(block.message.slot).BeaconBlock.hashTreeRoot(block.message));
-    if (!this.forkChoice.hasBlockHex(blockRoot)) {
+    const blockRoot = this.config.getForkTypes(block.message.slot).BeaconBlock.hashTreeRoot(block.message);
+    const blockRootHex = toHex(blockRoot);
+    if (!this.forkChoice.hasBlockHex(blockRootHex)) {
       blockToRemove.push(block);
 
       if (type === BlockInputType.postDeneb) {
-        blobsToRemove.push(blockInput.blobs);
-        this.db.blobsSidecar.remove(blockInput.blobs).catch((e) => {
-          this.logger.verbose("Error removing eagerly imported blobsSidecar", {blockRoot}, e);
-        });
+        const blobSidecars = blockInput.blobs;
+        blobsToRemove.push({blockRoot, slot: block.message.slot, blobSidecars});
       }
     }
   }
 
   await Promise.all([
     // TODO: Batch DB operations not with Promise.all but with level db ops
     this.db.block.batchRemove(blockToRemove),
-    this.db.blobsSidecar.batchRemove(blobsToRemove),
+    this.db.blobSidecars.batchRemove(blobsToRemove),
   ]);
 }

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

@@ -4,8 +4,8 @@ import {
   CachedBeaconStateAllForks,
   computeEpochAtSlot,
   computeStartSlotAtEpoch,
-  DataAvailableStatus,
   ExecutionPayloadStatus,
+  DataAvailableStatus,
   processSlots,
   stateTransition,
 } from "@lodestar/state-transition";

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

@@ -18,8 +18,6 @@ import {
   BackfilledRanges,
   BlobSidecarsRepository,
   BlobSidecarsArchiveRepository,
-  BlobsSidecarRepository,
-  BlobsSidecarArchiveRepository,
   BLSToExecutionChangeRepository,
 } from "./repositories/index.js";
 import {PreGenesisState, PreGenesisStateLastProcessedBlock} from "./single/index.js";
@@ -35,9 +33,6 @@ export class BeaconDb implements IBeaconDb {
 
   blobSidecars: BlobSidecarsRepository;
   blobSidecarsArchive: BlobSidecarsArchiveRepository;
-  // TODO DENEB: cleanup post full migration
-  blobsSidecar: BlobsSidecarRepository;
-  blobsSidecarArchive: BlobsSidecarArchiveRepository;
 
   stateArchive: StateArchiveRepository;
 
@@ -70,9 +65,6 @@ export class BeaconDb implements IBeaconDb {
 
     this.blobSidecars = new BlobSidecarsRepository(config, db);
     this.blobSidecarsArchive = new BlobSidecarsArchiveRepository(config, db);
-    // TODO DENEB: cleanup post full migration
-    this.blobsSidecar = new BlobsSidecarRepository(config, db);
-    this.blobsSidecarArchive = new BlobsSidecarArchiveRepository(config, db);
 
     this.stateArchive = new StateArchiveRepository(config, db);
     this.voluntaryExit = new VoluntaryExitRepository(config, db);
@@ -104,7 +96,7 @@ export class BeaconDb implements IBeaconDb {
 
   async pruneHotDb(): Promise<void> {
     // Prune all hot blobs
-    await this.blobsSidecar.batchDelete(await this.blobsSidecar.keys());
+    await this.blobSidecars.batchDelete(await this.blobSidecars.keys());
     // Prune all hot blocks
     // TODO: Enable once it's deemed safe
     // await this.block.batchDelete(await this.block.keys());