Enforce MAX_BLOBS_PER_BLOCK consensus rules

Author: Inphi

miner/worker.go

@@ -80,6 +80,7 @@ const (
 var (
 	errBlockInterruptedByNewHead  = errors.New("new head arrived while building block")
 	errBlockInterruptedByRecommit = errors.New("recommit interrupt while building block")
+	errMaxBlobsReached            = errors.New("reached max number of blobs per block")
 )
 
 // environment is the worker's current environment and holds all
@@ -98,6 +99,7 @@ type environment struct {
 	txs      []*types.Transaction
 	receipts []*types.Receipt
 	uncles   map[common.Hash]*types.Header
+	numBlobs int
 }
 
 // copy creates a deep copy of environment.
@@ -825,13 +827,19 @@ func (w *worker) updateSnapshot(env *environment) {
 func (w *worker) commitTransaction(env *environment, tx *types.Transaction) ([]*types.Log, error) {
 	snap := env.state.Snapshot()
 
+	txBlobCount := len(tx.DataHashes())
+	if env.numBlobs+txBlobCount > params.MaxBlobsPerBlock {
+		return nil, errMaxBlobsReached
+	}
+
 	receipt, err := core.ApplyTransaction(w.chainConfig, w.chain, &env.coinbase, env.gasPool, env.state, env.header, tx, &env.header.GasUsed, *w.chain.GetVMConfig())
 	if err != nil {
 		env.state.RevertToSnapshot(snap)
 		return nil, err
 	}
 	env.txs = append(env.txs, tx)
 	env.receipts = append(env.receipts, receipt)
+	env.numBlobs += txBlobCount
 
 	return receipt.Logs, nil
 }
@@ -919,6 +927,11 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP
 			log.Trace("Skipping unsupported transaction type", "sender", from, "type", tx.Type())
 			txs.Pop()
 
+		case errors.Is(err, errMaxBlobsReached):
+			// Shift, as the next tx from the account may not contain blobs
+			log.Trace("Skipping blob transaction. Reached max number of blobs in current context", "sender", from, "numBlobs", len(tx.DataHashes()))
+			txs.Shift()
+
 		default:
 			// Strange error, discard the transaction and get the next in line (note, the
 			// nonce-too-high clause will prevent us from executing in vain).

miner/worker_test.go

@@ -38,6 +38,8 @@ import (
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/params"
+	"github.com/holiman/uint256"
+	"github.com/protolambda/ztyp/view"
 )
 
 const (
@@ -198,6 +200,42 @@ func (b *testWorkerBackend) newRandomTx(creation bool) *types.Transaction {
 	return tx
 }
 
+func (b *testWorkerBackend) newRandomBlobTx(chainID *big.Int, nonce uint64) *types.Transaction {
+	var blobs types.Blobs
+	blobs = append(blobs, types.Blob{})
+
+	commitments, versionedHashes, aggregatedProof, err := blobs.ComputeCommitmentsAndAggregatedProof()
+	if err != nil {
+		panic(err)
+	}
+
+	gasPrice := big.NewInt(10 * params.InitialBaseFee).Uint64()
+	txData := &types.SignedBlobTx{
+		Message: types.BlobTxMessage{
+			ChainID:             view.Uint256View(*uint256.NewInt(chainID.Uint64())),
+			Nonce:               view.Uint64View(nonce),
+			Gas:                 view.Uint64View(testGas),
+			GasFeeCap:           view.Uint256View(*uint256.NewInt(gasPrice)),
+			GasTipCap:           view.Uint256View(*uint256.NewInt(gasPrice)),
+			Value:               view.Uint256View(*uint256.NewInt(1000)),
+			To:                  types.AddressOptionalSSZ{Address: (*types.AddressSSZ)(&testUserAddress)},
+			BlobVersionedHashes: versionedHashes,
+		},
+	}
+	wrapData := &types.BlobTxWrapData{
+		BlobKzgs:           commitments,
+		Blobs:              blobs,
+		KzgAggregatedProof: aggregatedProof,
+	}
+	tx := types.NewTx(txData, types.WithTxWrapData(wrapData))
+	signer := types.NewDankSigner(chainID)
+	tx, err = types.SignTx(tx, signer, testBankKey)
+	if err != nil {
+		panic(err)
+	}
+	return tx
+}
+
 func newTestWorker(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine, db ethdb.Database, blocks int) (*worker, *testWorkerBackend) {
 	backend := newTestWorkerBackend(t, chainConfig, engine, db, blocks)
 	backend.txPool.AddLocals(pendingTxs)
@@ -670,3 +708,85 @@ func testGetSealingWork(t *testing.T, chainConfig *params.ChainConfig, engine co
 		}
 	}
 }
+
+func testGenerateBlockWithBlobsAndImport(t *testing.T, isClique bool) {
+	var (
+		engine      consensus.Engine
+		chainConfig *params.ChainConfig
+		db          = rawdb.NewMemoryDatabase()
+	)
+	if isClique {
+		chainConfig = params.AllCliqueProtocolChanges
+		chainConfig.Clique = &params.CliqueConfig{Period: 1, Epoch: 30000}
+		engine = clique.New(chainConfig.Clique, db)
+	} else {
+		chainConfig = params.AllEthashProtocolChanges
+		engine = ethash.NewFaker()
+	}
+
+	chainConfig.LondonBlock = big.NewInt(0)
+	chainConfig.ShardingForkBlock = big.NewInt(0)
+	w, b := newTestWorker(t, chainConfig, engine, db, 0)
+	defer w.close()
+
+	// This test chain imports the mined blocks.
+	db2 := rawdb.NewMemoryDatabase()
+	b.genesis.MustCommit(db2)
+	chain, _ := core.NewBlockChain(db2, nil, b.chain.Config(), engine, vm.Config{}, nil, nil)
+	defer chain.Stop()
+
+	// Ignore empty commit here for less noise.
+	w.skipSealHook = func(task *task) bool {
+		return len(task.receipts) == 0
+	}
+
+	// Wait for mined blocks.
+	sub := w.mux.Subscribe(core.NewMinedBlockEvent{})
+	defer sub.Unsubscribe()
+
+	// Start mining!
+	w.start()
+
+	nonce := b.txPool.Nonce(testBankAddress)
+
+	var txs []*types.Transaction
+	for i := 0; i < params.MaxBlobsPerBlock+5; i++ {
+		txs = append(txs, b.newRandomBlobTx(chainConfig.ChainID, nonce))
+		nonce += 1
+	}
+	// Batch add blob txs to guarantee single block inclusion
+	errs := b.txPool.AddLocals(txs)
+	for i := range errs {
+		if errs[i] != nil {
+			panic(errs[i])
+		}
+	}
+
+	select {
+	case ev := <-sub.Chan():
+		block := ev.Data.(core.NewMinedBlockEvent).Block
+		if _, err := chain.InsertChain([]*types.Block{block}); err != nil {
+			t.Fatalf("failed to insert new mined block %d: %v", block.NumberU64(), err)
+		}
+		txs := block.Transactions()
+		var numBlobs int
+		for i := range txs {
+			numBlobs += len(txs[i].DataHashes())
+		}
+		// Assert that there are no more than params.MaxBlobsPerBlock blobs in the block
+		if numBlobs != params.MaxBlobsPerBlock {
+			t.Fatalf("unexpected number of blobs in block: %d", numBlobs)
+		}
+
+	case <-time.After(3 * time.Second): // Worker needs 1s to include new changes.
+		t.Fatalf("timeout")
+	}
+}
+
+func TestGenerateBlockWithBlobsAndImportEthash(t *testing.T) {
+	testGenerateBlockWithBlobsAndImport(t, false)
+}
+
+func TestGenerateBlockWithBlobsAndImportClique(t *testing.T) {
+	testGenerateBlockWithBlobsAndImport(t, true)
+}