feat: free the blobs (#5181)

* Free the blobs

fix the types

rejig the new constants in params

add comment for cleanup

update reqresp

fix api package

rename blobs repo

commit the wip modifications

further appropriate renaming

further references update

further reference updates

continue refac

fix reqresp build

further refac

further refac

fix api

fix db interface

fix beacondb alloc

build

fix api

improve blob verificaion

correct validation call

fixes

fix the produce block/blobs flow

reduce diff

blob gossip validation

update validations

cleanup block vali

reduce diff

handle gossip of block and blob

fix test for timebeing

modify publishing flow

fix import flow

onsidecarbyrange fix and some type fixes

fix sidecars by root

prune blockinput cache

fix kzg interface

small renaming

interface rename

fix fetch blockmaybeblobs by range test

fix build lint issues for now

c-kzg version fix

FullOrBlindedBlobSidecar changes

fix tests

complete the blob publishing flow

fix test

get the single node run functional

get the gossip blob flow working

fix peer syncing using req/resp

fix sidecar by root check

refactor blobsidecars hotdb and remove archive

add blob gossip validation flow

fix topic

fix the validation condition

add blob validation and test various sync modes

fix tests

rebase fixes

enable deneb spec tests

make blobsbyroot multi block

fixes

cleanup defunt builder endpoint

archive blobs post finalization uptill the blob window

serve finalized blobs within the blob prune window

fix test

fix test

lookup in archive as well

cleanup and improvements

rebase fx

Add 4844 sim test and override the field elements per blob

update image

add blob test

add test run in package

start unknown sync and range sync

finalize the sims

change the signing flow

fix test types

fix tests

fix test

lint

update tx type and corresponding ethereumjs image

update c-kzg and use blobs bundle proof

fix test

fix test

merge getblobsbundle into getpayloadv3

update images

fix genesis config

rebase fixes

fix test

update images

fix tests

lint

fix the sidecar request count limit

fix test

lint

rebase fixes

cleanup

fix unit tests

update kzg to big endian

devnet 6 integration

fix passing setup arg

update path

fix tests

fix blobs sidecar by range response

cleanup blob import vals

reduce diff

cleanups

disable blob sim tests till spec stablizes

handle blob bytes in blockinput

fixes

* apply feedback

* add todo for pre-emptive block/blobs pull

* rebase fix

.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());