client: add getBlobsV1 to the client to support CL blob import (#3711)

* client: add getBlobsV1 to the client to support CL blob import

* add the getBlobV1 spec and debug/fix the api

* debug and fix pending block spec and add assertions there as well

* apply feedback

* fix the lint issues

* spell checl

* refac the cache length param to fix build

* fix breasking tests

* improvs

* apply params fix for 4844 custom common

* fix spec

* 128 len check

* verify cache pruning in the test spec

* lint/spell

* fix typecheck

Author: g11tech

packages/client/src/config.ts

@@ -339,6 +339,11 @@ export interface ConfigOptions {
   startExecution?: boolean
   ignoreStatelessInvalidExecs?: boolean
 
+  /**
+   * The cache for blobs and proofs to support CL import blocks
+   */
+  blobsAndProofsCacheBlocks?: number
+
   /**
    * Enables Prometheus Metrics that can be collected for monitoring client health
    */
@@ -393,6 +398,9 @@ export class Config {
   // randomly kept it at 5 for fast testing purposes but ideally should be >=32 slots
   public static readonly SNAP_TRANSITION_SAFE_DEPTH = BigInt(5)
 
+  // support blobs and proofs cache for CL getBlobs for upto 1 epoch of data
+  public static readonly BLOBS_AND_PROOFS_CACHE_BLOCKS = 32
+
   public readonly logger: Logger
   public readonly syncmode: SyncMode
   public readonly vm?: VM
@@ -451,6 +459,8 @@ export class Config {
   public readonly startExecution: boolean
   public readonly ignoreStatelessInvalidExecs: boolean
 
+  public readonly blobsAndProofsCacheBlocks: number
+
   public synchronized: boolean
   public lastSynchronized?: boolean
   /** lastSyncDate in ms */
@@ -553,6 +563,9 @@ export class Config {
     this.chainCommon = common.copy()
     this.execCommon = common.copy()
 
+    this.blobsAndProofsCacheBlocks =
+      options.blobsAndProofsCacheBlocks ?? Config.BLOBS_AND_PROOFS_CACHE_BLOCKS
+
     this.discDns = this.getDnsDiscovery(options.discDns)
     this.discV4 = options.discV4 ?? true
 

packages/client/src/rpc/modules/engine/engine.ts

@@ -52,6 +52,7 @@ import type { Config } from '../../../config.js'
 import type { VMExecution } from '../../../execution/index.js'
 import type { FullEthereumService, Skeleton } from '../../../service/index.js'
 import type {
+  BlobAndProofV1,
   Bytes32,
   Bytes8,
   ExecutionPayloadBodyV1,
@@ -316,6 +317,13 @@ export class Engine {
       ]),
       () => this.connectionManager.updateStatus(),
     )
+
+    this.getBlobsV1 = cmMiddleware(
+      middleware(callWithStackTrace(this.getBlobsV1.bind(this), this._rpcDebug), 1, [
+        [validators.array(validators.bytes32)],
+      ]),
+      () => this.connectionManager.updateStatus(),
+    )
   }
 
   /**
@@ -1513,4 +1521,20 @@ export class Engine {
     }
     return payloads
   }
+
+  private async getBlobsV1(params: [[Bytes32]]): Promise<(BlobAndProofV1 | null)[]> {
+    if (params[0].length > 128) {
+      throw {
+        code: TOO_LARGE_REQUEST,
+        message: `More than 128 hashes queried`,
+      }
+    }
+
+    const blobsAndProof: (BlobAndProofV1 | null)[] = []
+    for (const versionedHashHex of params[0]) {
+      blobsAndProof.push(this.service.txPool.blobsAndProofsByHash.get(versionedHashHex) ?? null)
+    }
+
+    return blobsAndProof
+  }
 }

packages/client/src/rpc/modules/engine/types.ts

@@ -20,8 +20,7 @@ export enum Status {
 export type Bytes8 = PrefixedHexString
 export type Bytes20 = PrefixedHexString
 export type Bytes32 = PrefixedHexString
-// type Root = Bytes32
-export type Blob = Bytes32
+export type Blob = PrefixedHexString
 export type Bytes48 = PrefixedHexString
 export type Uint64 = PrefixedHexString
 export type Uint256 = PrefixedHexString
@@ -81,6 +80,11 @@ export type ExecutionPayloadBodyV1 = {
   withdrawals: WithdrawalV1[] | null
 }
 
+export type BlobAndProofV1 = {
+  blob: PrefixedHexString
+  proof: PrefixedHexString
+}
+
 export type ChainCache = {
   remoteBlocks: Map<String, Block>
   executedBlocks: Map<String, Block>

packages/client/src/service/txpool.ts

@@ -26,6 +26,7 @@ import type { PeerPool } from '../net/peerpool.js'
 import type { FullEthereumService } from './fullethereumservice.js'
 import type { Block } from '@ethereumjs/block'
 import type { FeeMarket1559Tx, LegacyTx, TypedTransaction } from '@ethereumjs/tx'
+import type { PrefixedHexString } from '@ethereumjs/util'
 import type { VM } from '@ethereumjs/vm'
 
 // Configuration constants
@@ -102,6 +103,10 @@ export class TxPool {
    * Maps an address to a `TxPoolObject`
    */
   public pool: Map<UnprefixedAddress, TxPoolObject[]>
+  public blobsAndProofsByHash: Map<
+    PrefixedHexString,
+    { blob: PrefixedHexString; proof: PrefixedHexString }
+  >
 
   /**
    * The number of txs currently in the pool
@@ -167,6 +172,10 @@ export class TxPool {
     this.service = options.service
 
     this.pool = new Map<UnprefixedAddress, TxPoolObject[]>()
+    this.blobsAndProofsByHash = new Map<
+      PrefixedHexString,
+      { blob: PrefixedHexString; proof: PrefixedHexString }
+    >()
     this.txsInPool = 0
     this.handled = new Map<UnprefixedHash, HandledObject>()
     this.knownByPeer = new Map<PeerId, SentObject[]>()
@@ -371,6 +380,16 @@ export class TxPool {
         this.config.metrics?.feeMarketEIP1559TxGauge?.inc()
       }
       if (isBlob4844Tx(tx)) {
+        // add to blobs and proofs cache
+        if (tx.blobs !== undefined && tx.kzgProofs !== undefined) {
+          for (const [i, versionedHash] of tx.blobVersionedHashes.entries()) {
+            const blob = tx.blobs![i]
+            const proof = tx.kzgProofs![i]
+            this.blobsAndProofsByHash.set(versionedHash, { blob, proof })
+          }
+          this.pruneBlobsAndProofsCache()
+        }
+
         this.config.metrics?.blobEIP4844TxGauge?.inc()
       }
     } catch (e) {
@@ -379,6 +398,24 @@ export class TxPool {
     }
   }
 
+  pruneBlobsAndProofsCache() {
+    const blobGasLimit = this.config.chainCommon.param('maxblobGasPerBlock')
+    const blobGasPerBlob = this.config.chainCommon.param('blobGasPerBlob')
+    const allowedBlobsPerBlock = Number(blobGasLimit / blobGasPerBlob)
+
+    const pruneLength =
+      this.blobsAndProofsByHash.size - allowedBlobsPerBlock * this.config.blobsAndProofsCacheBlocks
+    let pruned = 0
+    // since keys() is sorted by insertion order this prunes the oldest data in cache
+    for (const versionedHash of this.blobsAndProofsByHash.keys()) {
+      if (pruned >= pruneLength) {
+        break
+      }
+      this.blobsAndProofsByHash.delete(versionedHash)
+      pruned++
+    }
+  }
+
   /**
    * Returns the available txs from the pool
    * @param txHashes

packages/client/test/miner/pendingBlock.spec.ts

@@ -12,6 +12,7 @@ import {
   commitmentsToVersionedHashes,
   getBlobs,
   hexToBytes,
+  intToHex,
   randomBytes,
 } from '@ethereumjs/util'
 import { createVM } from '@ethereumjs/vm'
@@ -28,7 +29,9 @@ import { mockBlockchain } from '../rpc/mockBlockchain.js'
 
 import type { Blockchain } from '@ethereumjs/blockchain'
 import type { TypedTransaction } from '@ethereumjs/tx'
+import type { PrefixedHexString } from '@ethereumjs/util'
 import type { VM } from '@ethereumjs/vm'
+
 const kzg = new microEthKZG(trustedSetup)
 
 const A = {
@@ -353,24 +356,48 @@ describe('[PendingBlock]', async () => {
     })
 
     const { txPool } = setup()
+    txPool['config'].chainCommon.setHardfork(Hardfork.Cancun)
+
+    // fill up the blobsAndProofsByHash and proofs cache before adding a blob tx
+    // for cache pruning check
+    const fillBlobs = getBlobs('hello world')
+    const fillCommitments = blobsToCommitments(kzg, fillBlobs)
+    const fillProofs = blobsToProofs(kzg, fillBlobs, fillCommitments)
+    const fillBlobAndProof = { blob: fillBlobs[0], proof: fillProofs[0] }
+
+    const blobGasLimit = txPool['config'].chainCommon.param('maxblobGasPerBlock')
+    const blobGasPerBlob = txPool['config'].chainCommon.param('blobGasPerBlob')
+    const allowedBlobsPerBlock = Number(blobGasLimit / blobGasPerBlob)
+    const allowedLength = allowedBlobsPerBlock * txPool['config'].blobsAndProofsCacheBlocks
+
+    for (let i = 0; i < allowedLength; i++) {
+      // this is space efficient as same object is inserted in dummy positions
+      txPool.blobsAndProofsByHash.set(intToHex(i), fillBlobAndProof)
+    }
+    assert.equal(txPool.blobsAndProofsByHash.size, allowedLength, 'fill the cache to capacity')
 
-    const blobs = getBlobs('hello world')
-    const commitments = blobsToCommitments(kzg, blobs)
-    const blobVersionedHashes = commitmentsToVersionedHashes(commitments)
-    const proofs = blobsToProofs(kzg, blobs, commitments)
-
-    // Create 3 txs with 2 blobs each so that only 2 of them can be included in a build
+    // Create 2 txs with 3 blobs each so that only 2 of them can be included in a build
+    let blobs: PrefixedHexString[] = [],
+      proofs: PrefixedHexString[] = [],
+      versionedHashes: PrefixedHexString[] = []
     for (let x = 0; x <= 2; x++) {
+      // generate unique blobs different from fillBlobs
+      const txBlobs = [
+        ...getBlobs(`hello world-${x}1`),
+        ...getBlobs(`hello world-${x}2`),
+        ...getBlobs(`hello world-${x}3`),
+      ]
+      assert.equal(txBlobs.length, 3, '3 blobs should be created')
+      const txCommitments = blobsToCommitments(kzg, txBlobs)
+      const txBlobVersionedHashes = commitmentsToVersionedHashes(txCommitments)
+      const txProofs = blobsToProofs(kzg, txBlobs, txCommitments)
+
       const txA01 = createBlob4844Tx(
         {
-          blobVersionedHashes: [
-            ...blobVersionedHashes,
-            ...blobVersionedHashes,
-            ...blobVersionedHashes,
-          ],
-          blobs: [...blobs, ...blobs, ...blobs],
-          kzgCommitments: [...commitments, ...commitments, ...commitments],
-          kzgProofs: [...proofs, ...proofs, ...proofs],
+          blobVersionedHashes: txBlobVersionedHashes,
+          blobs: txBlobs,
+          kzgCommitments: txCommitments,
+          kzgProofs: txProofs,
           maxFeePerBlobGas: 100000000n,
           gasLimit: 0xffffffn,
           maxFeePerGas: 1000000000n,
@@ -381,6 +408,30 @@ describe('[PendingBlock]', async () => {
         { common },
       ).sign(A.privateKey)
       await txPool.add(txA01)
+
+      // accumulate for verification
+      blobs = [...blobs, ...txBlobs]
+      proofs = [...proofs, ...txProofs]
+      versionedHashes = [...versionedHashes, ...txBlobVersionedHashes]
+    }
+
+    assert.equal(
+      txPool.blobsAndProofsByHash.size,
+      allowedLength,
+      'cache should be prune and stay at same size',
+    )
+    // check if blobs and proofs are added in txpool by versioned hashes
+    for (let i = 0; i < versionedHashes.length; i++) {
+      const versionedHash = versionedHashes[i]
+      const blob = blobs[i]
+      const proof = proofs[i]
+
+      const blobAndProof = txPool.blobsAndProofsByHash.get(versionedHash) ?? {
+        blob: '0x0',
+        proof: '0x0',
+      }
+      assert.equal(blob, blobAndProof.blob, 'blob should match')
+      assert.equal(proof, blobAndProof.proof, 'proof should match')
     }
 
     // Add one other normal tx for nonce 3 which should also be not included in the build

packages/client/test/rpc/engine/getPayloadV3.spec.ts

@@ -110,6 +110,16 @@ describe(method, () => {
     ).sign(pkey)
 
     await service.txPool.add(tx, true)
+
+    // check the blob and proof is available via getBlobsV1
+    res = await rpc.request('engine_getBlobsV1', [txVersionedHashes])
+    const blobsAndProofs = res.result
+    for (let i = 0; i < txVersionedHashes.length; i++) {
+      const { blob, proof } = blobsAndProofs[i]
+      assert.equal(blob, txBlobs[i])
+      assert.equal(proof, txProofs[i])
+    }
+
     res = await rpc.request('engine_getPayloadV3', [payloadId])
 
     const { executionPayload, blobsBundle } = res.result

packages/client/test/rpc/eth/sendRawTransaction.spec.ts

@@ -1,4 +1,4 @@
-import { BlockHeader } from '@ethereumjs/block'
+import { BlockHeader, paramsBlock } from '@ethereumjs/block'
 import { Common, Hardfork, Mainnet, createCommonFromGethGenesis } from '@ethereumjs/common'
 import { MerkleStateManager } from '@ethereumjs/statemanager'
 import { createBlob4844Tx, createFeeMarket1559TxFromRLP, createLegacyTx } from '@ethereumjs/tx'
@@ -228,7 +228,9 @@ describe(method, () => {
       chain: 'customChain',
       hardfork: Hardfork.Cancun,
       customCrypto: { kzg },
+      params: paramsBlock,
     })
+
     common.setHardfork(Hardfork.Cancun)
     const { rpc, client } = await baseSetup({
       commonChain: common,

packages/client/test/rpc/helpers.ts

@@ -1,4 +1,4 @@
-import { createBlockHeader } from '@ethereumjs/block'
+import { createBlockHeader, paramsBlock } from '@ethereumjs/block'
 import { createBlockchain } from '@ethereumjs/blockchain'
 import {
   Common,
@@ -235,6 +235,7 @@ export async function setupChain(genesisFile: any, chainName = 'dev', clientOpts
   const common = createCommonFromGethGenesis(genesisFile, {
     chain: chainName,
     customCrypto: clientOpts.customCrypto,
+    params: paramsBlock,
   })
   common.setHardforkBy({
     blockNumber: 0,