client: Limit blobs as per the maxDataGasPerBlock for block building

packages/client/lib/miner/miner.ts

@@ -293,7 +293,7 @@ export class Miner {
       },
     })
 
-    const txs = await this.service.txPool.txsByPriceAndNonce(vmCopy, baseFeePerGas)
+    const txs = await this.service.txPool.txsByPriceAndNonce(vmCopy, { baseFee: baseFeePerGas })
     this.config.logger.info(
       `Miner: Assembling block from ${txs.length} eligible txs ${
         typeof baseFeePerGas === 'bigint' && baseFeePerGas !== BigInt(0)

packages/client/lib/miner/pendingBlock.ts

@@ -148,8 +148,20 @@ export class PendingBlock {
 
     this.pendingPayloads.set(payloadId, builder)
 
+    // Get if and how many blobs are allowed in the tx
+    let allowedBlobs
+    if (vm._common.isActivatedEIP(4844)) {
+      const dataGasLimit = vm._common.param('gasConfig', 'maxDataGasPerBlock')
+      const dataGasPerBlob = vm._common.param('gasConfig', 'dataGasPerBlob')
+      allowedBlobs = Number(dataGasLimit / dataGasPerBlob)
+    } else {
+      allowedBlobs = 0
+    }
     // Add current txs in pool
-    const txs = await this.txPool.txsByPriceAndNonce(vm, baseFeePerGas)
+    const txs = await this.txPool.txsByPriceAndNonce(vm, {
+      baseFee: baseFeePerGas,
+      allowedBlobs,
+    })
     this.config.logger.info(
       `Pending: Assembling block from ${txs.length} eligible txs (baseFee: ${baseFeePerGas})`
     )
@@ -225,10 +237,26 @@ export class PendingBlock {
     if (blockStatus.status === BuildStatus.Build) {
       return [blockStatus.block, builder.transactionReceipts, builder.minerValue]
     }
-    const { vm, headerData } = builder as any
+    const { vm, headerData } = builder as unknown as { vm: VM; headerData: HeaderData }
+
+    // get the number of blobs that can be further added
+    let allowedBlobs
+    if (vm._common.isActivatedEIP(4844)) {
+      const bundle = this.blobsBundles.get(payloadId) ?? { blobs: [], commitments: [], proofs: [] }
+      const dataGasLimit = vm._common.param('gasConfig', 'maxDataGasPerBlock')
+      const dataGasPerBlob = vm._common.param('gasConfig', 'dataGasPerBlob')
+      allowedBlobs = Number(dataGasLimit / dataGasPerBlob) - bundle.blobs.length
+    } else {
+      allowedBlobs = 0
+    }
 
     // Add new txs that the pool received
-    const txs = (await this.txPool.txsByPriceAndNonce(vm, headerData.baseFeePerGas)).filter(
+    const txs = (
+      await this.txPool.txsByPriceAndNonce(vm, {
+        baseFee: headerData.baseFeePerGas! as bigint,
+        allowedBlobs,
+      })
+    ).filter(
       (tx) =>
         (builder as any).transactions.some((t: TypedTransaction) =>
           equalsBytes(t.hash(), tx.hash())
@@ -281,18 +309,21 @@ export class PendingBlock {
     }
 
     const block = await builder.build()
+    // Construct blobs bundle
+    const blobs = block._common.isActivatedEIP(4844)
+      ? this.constructBlobsBundle(payloadId, blobTxs)
+      : undefined
+
     const withdrawalsStr = block.withdrawals ? ` withdrawals=${block.withdrawals.length}` : ''
+    const blobsStr = blobs ? ` blobs=${blobs.blobs.length}` : ''
     this.config.logger.info(
       `Pending: Built block number=${block.header.number} txs=${
         block.transactions.length
-      }${withdrawalsStr} skippedByAddErrors=${skippedByAddErrors}  hash=${bytesToHex(block.hash())}`
+      }${withdrawalsStr}${blobsStr} skippedByAddErrors=${skippedByAddErrors}  hash=${bytesToHex(
+        block.hash()
+      )}`
     )
 
-    // Construct blobs bundle
-    const blobs = block._common.isActivatedEIP(4844)
-      ? this.constructBlobsBundle(payloadId, blobTxs)
-      : undefined
-
     return [block, builder.transactionReceipts, builder.minerValue, blobs]
   }
 

packages/client/lib/rpc/modules/eth.ts

@@ -924,6 +924,17 @@ export class Eth {
       if (txBuf[0] === 0x03) {
         // Blob Transactions sent over RPC are expected to be in Network Wrapper format
         tx = BlobEIP4844Transaction.fromSerializedBlobTxNetworkWrapper(txBuf, { common })
+
+        const dataGasLimit = common.param('gasConfig', 'maxDataGasPerBlock')
+        const dataGasPerBlob = common.param('gasConfig', 'dataGasPerBlob')
+
+        if (BigInt((tx.blobs ?? []).length) * dataGasPerBlob > dataGasLimit) {
+          throw Error(
+            `tx blobs=${(tx.blobs ?? []).length} exceeds block limit=${
+              dataGasLimit / dataGasPerBlob
+            }`
+          )
+        }
       } else {
         tx = TransactionFactory.fromSerializedData(txBuf, { common })
       }

packages/client/lib/service/txpool.ts

@@ -687,11 +687,14 @@ export class TxPool {
    *
    * @param baseFee Provide a baseFee to exclude txs with a lower gasPrice
    */
-  async txsByPriceAndNonce(vm: VM, baseFee?: bigint) {
+  async txsByPriceAndNonce(
+    vm: VM,
+    { baseFee, allowedBlobs }: { baseFee?: bigint; allowedBlobs?: number } = {}
+  ) {
     const txs: TypedTransaction[] = []
     // Separate the transactions by account and sort by nonce
     const byNonce = new Map<string, TypedTransaction[]>()
-    const skippedStats = { byNonce: 0, byPrice: 0 }
+    const skippedStats = { byNonce: 0, byPrice: 0, byBlobsLimit: 0 }
     for (const [address, poolObjects] of this.pool) {
       let txsSortedByNonce = poolObjects
         .map((obj) => obj.tx)
@@ -729,22 +732,43 @@ export class TxPool {
       byNonce.set(address, txs.slice(1))
     }
     // Merge by replacing the best with the next from the same account
+    let blobsCount = 0
     while (byPrice.length > 0) {
       // Retrieve the next best transaction by price
       const best = byPrice.remove()
       if (!best) break
+
       // Push in its place the next transaction from the same account
       const address = best.getSenderAddress().toString().slice(2)
       const accTxs = byNonce.get(address)!
-      if (accTxs.length > 0) {
-        byPrice.insert(accTxs[0])
-        byNonce.set(address, accTxs.slice(1))
+
+      // Insert the best tx into byPrice if
+      //   i) this is not a blob tx,
+      //   ii) or there is no blobs limit provided
+      //   iii) or blobs are still within limit if this best tx's blobs are included
+      if (
+        !(best instanceof BlobEIP4844Transaction) ||
+        allowedBlobs === undefined ||
+        ((best as BlobEIP4844Transaction).blobs ?? []).length + blobsCount <= allowedBlobs
+      ) {
+        if (accTxs.length > 0) {
+          byPrice.insert(accTxs[0])
+          byNonce.set(address, accTxs.slice(1))
+        }
+        // Accumulate the best priced transaction and increment blobs count
+        txs.push(best)
+        if (best instanceof BlobEIP4844Transaction) {
+          blobsCount += ((best as BlobEIP4844Transaction).blobs ?? []).length
+        }
+      } else {
+        // Since no more blobs can fit in the block, not only skip inserting in byPrice but also remove all other
+        // txs (blobs or not) of this sender address from further consideration
+        skippedStats.byBlobsLimit += 1 + accTxs.length
+        byNonce.set(address, [])
       }
-      // Accumulate the best priced transaction
-      txs.push(best)
     }
     this.config.logger.info(
-      `txsByPriceAndNonce selected txs=${txs.length}, skipped byNonce=${skippedStats.byNonce} byPrice=${skippedStats.byPrice}`
+      `txsByPriceAndNonce selected txs=${txs.length}, skipped byNonce=${skippedStats.byNonce} byPrice=${skippedStats.byPrice} byBlobsLimit=${skippedStats.byBlobsLimit}`
     )
     return txs
   }

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

@@ -1,6 +1,6 @@
 import { Block, BlockHeader } from '@ethereumjs/block'
 import { Common, Chain as CommonChain, Hardfork } from '@ethereumjs/common'
-import { BlobEIP4844Transaction, Transaction } from '@ethereumjs/tx'
+import { BlobEIP4844Transaction, FeeMarketEIP1559Transaction, Transaction } from '@ethereumjs/tx'
 import {
   Account,
   Address,
@@ -275,38 +275,66 @@ tape('[PendingBlock]', async (t) => {
       hardfork: Hardfork.Cancun,
     })
     const { txPool } = setup()
+
     const blobs = getBlobs('hello world')
     const commitments = blobsToCommitments(blobs)
     const versionedHashes = commitmentsToVersionedHashes(commitments)
     const proofs = blobsToProofs(blobs, commitments)
 
-    const txA01 = BlobEIP4844Transaction.fromTxData(
+    // Create 3 txs with 2 blobs each so that only 2 of them can be included in a build
+    for (let x = 0; x <= 2; x++) {
+      const txA01 = BlobEIP4844Transaction.fromTxData(
+        {
+          versionedHashes,
+          blobs: [...blobs, ...blobs],
+          kzgCommitments: [...commitments, ...commitments],
+          kzgProofs: [...proofs, ...proofs],
+          maxFeePerDataGas: 100000000n,
+          gasLimit: 0xffffffn,
+          maxFeePerGas: 1000000000n,
+          maxPriorityFeePerGas: 100000000n,
+          to: randomBytes(20),
+          nonce: BigInt(x),
+        },
+        { common }
+      ).sign(A.privateKey)
+      await txPool.add(txA01)
+    }
+
+    // Add one other normal tx for nonce 3 which should also be not included in the build
+    const txNorm = FeeMarketEIP1559Transaction.fromTxData(
       {
-        versionedHashes,
-        blobs,
-        kzgCommitments: commitments,
-        kzgProofs: proofs,
-        maxFeePerDataGas: 100000000n,
         gasLimit: 0xffffffn,
         maxFeePerGas: 1000000000n,
         maxPriorityFeePerGas: 100000000n,
         to: randomBytes(20),
+        nonce: BigInt(3),
       },
       { common }
     ).sign(A.privateKey)
-    await txPool.add(txA01)
+    await txPool.add(txNorm)
+    st.equal(txPool.txsInPool, 4, '4 txs should still be in the pool')
+
     const pendingBlock = new PendingBlock({ config, txPool })
     const vm = await VM.create({ common })
-    await setBalance(vm, A.address, BigInt(5000000000000000))
+    await setBalance(vm, A.address, BigInt(500000000000000000))
     const parentBlock = await vm.blockchain.getCanonicalHeadBlock!()
+    // stub the vm's common set hf to do nothing but stay in cancun
+    vm._common.setHardforkByBlockNumber = (_a: bigint, _b?: bigint, _c?: bigint) => {
+      return vm._common.hardfork()
+    }
     const payloadId = await pendingBlock.start(vm, parentBlock)
-    await pendingBlock.build(payloadId)
+    const [block, _receipts, _value, blobsBundles] = (await pendingBlock.build(payloadId)) ?? []
+
+    st.ok(block !== undefined && blobsBundles !== undefined)
+    st.equal(block!.transactions.length, 2, 'Only two blob txs should be included')
+    st.equal(blobsBundles!.blobs.length, 4, 'maximum 4 blobs should be included')
+    st.equal(blobsBundles!.commitments.length, 4, 'maximum 4 commitments should be included')
+    st.equal(blobsBundles!.proofs.length, 4, 'maximum 4 proofs should be included')
 
-    const blobsBundles = pendingBlock.blobsBundles.get(bytesToPrefixedHexString(payloadId))!
-    st.ok(blobsBundles !== undefined)
-    const pendingBlob = blobsBundles.blobs[0]
+    const pendingBlob = blobsBundles!.blobs[0]
     st.ok(pendingBlob !== undefined && equalsBytes(pendingBlob, blobs[0]))
-    const blobProof = blobsBundles.proofs[0]
+    const blobProof = blobsBundles!.proofs[0]
     st.ok(blobProof !== undefined && equalsBytes(blobProof, proofs[0]))
     st.end()
   })