all: 4844 devnet 6 integration

Author: MariusVanDerWijden

consensus/beacon/consensus.go

@@ -379,6 +379,14 @@ func (beacon *Beacon) FinalizeAndAssemble(chain consensus.ChainHeaderReader, hea
 			return nil, errors.New("withdrawals set before Shanghai activation")
 		}
 	}
+	if chain.Config().IsCancun(header.Number, header.Time) {
+		var blobs int
+		for _, tx := range txs {
+			blobs += len(tx.BlobHashes())
+		}
+		blobGasUsed := uint64(blobs * params.BlobTxBlobGasPerBlob)
+		header.BlobGasUsed = &blobGasUsed
+	}
 	// Finalize and assemble the block.
 	beacon.Finalize(chain, header, state, txs, uncles, withdrawals)
 

consensus/misc/eip4844/eip4844_test.go

@@ -45,8 +45,8 @@ func TestCalcExcessBlobGas(t *testing.T) {
 		// The excess blob gas should decrease by however much the target was
 		// under-shot, capped at zero.
 		{params.BlobTxTargetBlobGasPerBlock, params.BlobTxTargetBlobGasPerBlock / params.BlobTxBlobGasPerBlob, params.BlobTxTargetBlobGasPerBlock},
-		{params.BlobTxTargetBlobGasPerBlock, (params.BlobTxTargetBlobGasPerBlock / params.BlobTxBlobGasPerBlob) - 1, params.BlobTxBlobGasPerBlob},
-		{params.BlobTxTargetBlobGasPerBlock, (params.BlobTxTargetBlobGasPerBlock / params.BlobTxBlobGasPerBlob) - 2, 0},
+		{params.BlobTxTargetBlobGasPerBlock, (params.BlobTxTargetBlobGasPerBlock / params.BlobTxBlobGasPerBlob) - 1, 2 * params.BlobTxBlobGasPerBlob},
+		{params.BlobTxTargetBlobGasPerBlock, (params.BlobTxTargetBlobGasPerBlock / params.BlobTxBlobGasPerBlob) - 2, params.BlobTxBlobGasPerBlob},
 		{params.BlobTxBlobGasPerBlob - 1, (params.BlobTxTargetBlobGasPerBlock / params.BlobTxBlobGasPerBlob) - 1, 0},
 	}
 	for _, tt := range tests {
@@ -64,8 +64,9 @@ func TestCalcBlobFee(t *testing.T) {
 	}{
 		{0, 1},
 		{1542706, 1},
-		{1542707, 2},
-		{10 * 1024 * 1024, 111},
+		{1542707, 1},
+		{3085414, 2},
+		{10 * 1024 * 1024, 23},
 	}
 	for i, tt := range tests {
 		have := CalcBlobFee(tt.excessBlobGas)

core/genesis.go

@@ -496,6 +496,14 @@ func (g *Genesis) ToBlock() *types.Block {
 			}
 		}
 	}
+	if g.Config != nil && g.Config.IsCancun(big.NewInt(int64(g.Number)), g.Timestamp) {
+		if g.ExcessBlobGas == nil {
+			var excessBlobGas uint64
+			head.ExcessBlobGas = &excessBlobGas
+		} else {
+			head.ExcessBlobGas = g.ExcessBlobGas
+		}
+	}
 	return types.NewBlock(head, nil, nil, nil, trie.NewStackTrie(nil)).WithWithdrawals(withdrawals)
 }
 

core/txpool/blobpool/blobpool.go

@@ -954,6 +954,8 @@ func (p *BlobPool) reinject(addr common.Address, txhash common.Hash) {
 	}
 	p.lookup[meta.hash] = meta.id
 	p.stored += uint64(meta.size)
+
+	p.eventFeed.Send(core.NewTxsEvent{Txs: types.Transactions{tx}})
 }
 
 // SetGasTip implements txpool.SubPool, allowing the blob pool's gas requirements
@@ -1036,7 +1038,6 @@ func (p *BlobPool) validateTx(tx *types.Transaction) error {
 	// Ensure the transaction adheres to the stateful pool filters (nonce, balance)
 	stateOpts := &txpool.ValidationOptionsWithState{
 		State: p.state,
-
 		FirstNonceGap: func(addr common.Address) uint64 {
 			// Nonce gaps are not permitted in the blob pool, the first gap will
 			// be the next nonce shifted by however many transactions we already
@@ -1297,6 +1298,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) {
 	}
 	p.updateStorageMetrics()
 
+	p.eventFeed.Send(core.NewTxsEvent{Txs: types.Transactions{tx}})
 	return nil
 }
 

core/txpool/blobpool/slotter_test.go

@@ -46,6 +46,8 @@ func TestNewSlotter(t *testing.T) {
 		10*blobSize + txAvgSize, // 1-4 blobs + unexpectedly large tx infos < 4 blobs + max tx metadata size
 		11*blobSize + txAvgSize, // 1-4 blobs + unexpectedly large tx infos < 4 blobs + max tx metadata size
 		12*blobSize + txAvgSize, // 1-4 blobs + unexpectedly large tx infos >= 4 blobs + max tx metadata size
+		13*blobSize + txAvgSize, // 1-4 blobs + unexpectedly large tx infos >= 4 blobs + max tx metadata size
+		14*blobSize + txAvgSize, // 1-4 blobs + unexpectedly large tx infos >= 4 blobs + max tx metadata size
 	}
 	if len(shelves) != len(want) {
 		t.Errorf("shelves count mismatch: have %d, want %d", len(shelves), len(want))

core/types/types_test.go

@@ -1287,7 +1287,11 @@ func testTransactionFetcher(t *testing.T, tt txFetcherTest) {
 			}
 
 		case doTxEnqueue:
-			if err := fetcher.Enqueue(step.peer, step.txs, step.direct); err != nil {
+			var txs []*types.BlobTxWithBlobs
+			for _, tx := range step.txs {
+				txs = append(txs, types.NewBlobTxWithBlobs(tx, nil, nil, nil))
+			}
+			if err := fetcher.Enqueue(step.peer, txs, step.direct); err != nil {
 				t.Errorf("step %d: %v", i, err)
 			}
 			<-wait // Fetcher needs to process this, wait until it's done

eth/fetcher/tx_fetcher_test.go

@@ -625,7 +625,7 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) {
 		peers := h.peers.peersWithoutTransaction(tx.Hash())
 
 		var numDirect int
-		if tx.Size() <= txMaxBroadcastSize {
+		if tx.Size() <= txMaxBroadcastSize && tx.Type() != types.BlobTxType {
 			numDirect = int(math.Sqrt(float64(len(peers))))
 		}
 		// Send the tx unconditionally to a subset of our peers

eth/handler.go

@@ -30,12 +30,14 @@ import (
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/params"
+	"github.com/holiman/uint256"
 )
 
 // testEthHandler is a mock event handler to listen for inbound network requests
@@ -71,7 +73,11 @@ func (h *testEthHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
 		return nil
 
 	case *eth.PooledTransactionsPacket:
-		h.txBroadcasts.Send(([]*types.Transaction)(*packet))
+		var txs []*types.Transaction
+		for _, tx := range *packet {
+			txs = append(txs, tx.Transaction)
+		}
+		h.txBroadcasts.Send(txs)
 		return nil
 
 	default:
@@ -454,6 +460,87 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
 	}
 }
 
+func TestBlobTransactionPropagation68(t *testing.T) { testBlobTransactionPropagation(t, eth.ETH68) }
+
+func testBlobTransactionPropagation(t *testing.T, protocol uint) {
+	t.Parallel()
+
+	// Create a source handler to send transactions from and a number of sinks
+	// to receive them. We need multiple sinks since a one-to-one peering would
+	// broadcast all transactions without announcement.
+	source := newTestHandler()
+	source.handler.snapSync.Store(false) // Avoid requiring snap, otherwise some will be dropped below
+	defer source.close()
+
+	sinks := make([]*testHandler, 10)
+	for i := 0; i < len(sinks); i++ {
+		sinks[i] = newTestHandler()
+		defer sinks[i].close()
+
+		sinks[i].handler.acceptTxs.Store(true) // mark synced to accept transactions
+	}
+	// Interconnect all the sink handlers with the source handler
+	for i, sink := range sinks {
+		sink := sink // Closure for gorotuine below
+
+		sourcePipe, sinkPipe := p2p.MsgPipe()
+		defer sourcePipe.Close()
+		defer sinkPipe.Close()
+
+		sourcePeer := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{byte(i + 1)}, "", nil, sourcePipe), sourcePipe, source.txpool)
+		sinkPeer := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{0}, "", nil, sinkPipe), sinkPipe, sink.txpool)
+		defer sourcePeer.Close()
+		defer sinkPeer.Close()
+
+		go source.handler.runEthPeer(sourcePeer, func(peer *eth.Peer) error {
+			return eth.Handle((*ethHandler)(source.handler), peer)
+		})
+		go sink.handler.runEthPeer(sinkPeer, func(peer *eth.Peer) error {
+			return eth.Handle((*ethHandler)(sink.handler), peer)
+		})
+	}
+	// Subscribe to all the transaction pools
+	txChs := make([]chan core.NewTxsEvent, len(sinks))
+	for i := 0; i < len(sinks); i++ {
+		txChs[i] = make(chan core.NewTxsEvent, 1024)
+
+		sub := sinks[i].txpool.SubscribeNewTxsEvent(txChs[i])
+		defer sub.Unsubscribe()
+	}
+	// Fill the source pool with transactions and wait for them at the sinks
+	txs := make([]*txpool.Transaction, 1024)
+	for nonce := range txs {
+		tx := types.NewTx(&types.BlobTx{
+			Nonce:      uint64(nonce),
+			To:         common.Address{},
+			GasTipCap:  uint256.NewInt(0),
+			GasFeeCap:  uint256.NewInt(0),
+			Gas:        100000,
+			Value:      uint256.NewInt(0),
+			BlobHashes: make([]common.Hash, 1),
+			ChainID:    uint256.NewInt(0),
+			Data:       nil,
+		})
+		tx, _ = types.SignTx(tx, types.NewCancunSigner(common.Big0), testKey)
+
+		txs[nonce] = &txpool.Transaction{Tx: tx, BlobTxBlobs: make([]kzg4844.Blob, 1), BlobTxCommits: make([]kzg4844.Commitment, 1), BlobTxProofs: make([]kzg4844.Proof, 1)}
+	}
+	source.txpool.Add(txs, false, false)
+
+	// Iterate through all the sinks and ensure they all got the transactions
+	for i := range sinks {
+		for arrived, timeout := 0, false; arrived < len(txs) && !timeout; {
+			select {
+			case event := <-txChs[i]:
+				arrived += len(event.Txs)
+			case <-time.After(2 * time.Second):
+				t.Errorf("sink %d: transaction propagation timed out: have %d, want %d", i, arrived, len(txs))
+				timeout = true
+			}
+		}
+	}
+}
+
 // Tests that blocks are broadcast to a sqrt number of peers only.
 func TestBroadcastBlock1Peer(t *testing.T)    { testBroadcastBlock(t, 1, 1) }
 func TestBroadcastBlock2Peers(t *testing.T)   { testBroadcastBlock(t, 2, 1) }

eth/handler_eth_test.go

@@ -80,6 +80,7 @@ func (p *Peer) broadcastTransactions() {
 				size        common.StorageSize
 			)
 			for i := 0; i < len(queue) && size < maxTxPacketSize; i++ {
+				// Do not broadcast blob transactions
 				if tx := p.txpool.Get(queue[i]); tx != nil {
 					txs = append(txs, tx)
 					size += common.StorageSize(tx.Size())