Blobtx mining + brodcasting

core/types/tx_blob.go

@@ -54,6 +54,15 @@ type BlobTx struct {
 
 type BlobTxSidecars []*BlobTxSidecar
 
+// Len returns the length of s.
+func (s BlobTxSidecars) Len() int { return len(s) }
+
+// EncodeIndex encodes the i'th BlobTxSidecar to w. Note that this does not check for errors
+// because we assume that BlobTxSidecars will only ever contain valid sidecars
+func (s BlobTxSidecars) EncodeIndex(i int, w *bytes.Buffer) {
+	rlp.Encode(w, s[i])
+}
+
 // BlobTxSidecar contains the blobs of a blob transaction.
 type BlobTxSidecar struct {
 	Blobs       []kzg4844.Blob       // Blobs needed by the blob pool

eth/downloader/downloader.go

@@ -1388,7 +1388,7 @@ func (d *Downloader) importBlockResults(results []*fetchResult) error {
 	)
 	blocks := make([]*types.Block, len(results))
 	for i, result := range results {
-		blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles).WithWithdrawals(result.Withdrawals)
+		blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles).WithWithdrawals(result.Withdrawals).WithBlobs(result.Sidecars)
 	}
 	// Downloaded blocks are always regarded as trusted after the
 	// transition. Because the downloaded chain is guided by the
@@ -1599,7 +1599,7 @@ func (d *Downloader) commitSnapSyncData(results []*fetchResult, stateSync *state
 	blocks := make([]*types.Block, len(results))
 	receipts := make([]types.Receipts, len(results))
 	for i, result := range results {
-		blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles).WithWithdrawals(result.Withdrawals)
+		blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles).WithWithdrawals(result.Withdrawals).WithBlobs(result.Sidecars)
 		receipts[i] = result.Receipts
 	}
 	if index, err := d.blockchain.InsertReceiptChain(blocks, receipts, d.ancientLimit); err != nil {
@@ -1610,7 +1610,7 @@ func (d *Downloader) commitSnapSyncData(results []*fetchResult, stateSync *state
 }
 
 func (d *Downloader) commitPivotBlock(result *fetchResult) error {
-	block := types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles).WithWithdrawals(result.Withdrawals)
+	block := types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles).WithWithdrawals(result.Withdrawals).WithBlobs(result.Sidecars)
 	log.Debug("Committing snap sync pivot as new head", "number", block.Number(), "hash", block.Hash())
 
 	// Commit the pivot block as the new head, will require full sync from here on

eth/downloader/fetchers_concurrent_bodies.go

@@ -89,10 +89,10 @@ func (q *bodyQueue) request(peer *peerConnection, req *fetchRequest, resCh chan
 // deliver is responsible for taking a generic response packet from the concurrent
 // fetcher, unpacking the body data and delivering it to the downloader's queue.
 func (q *bodyQueue) deliver(peer *peerConnection, packet *eth.Response) (int, error) {
-	txs, uncles, withdrawals := packet.Res.(*eth.BlockBodiesResponse).Unpack()
+	txs, uncles, withdrawals, sidecars := packet.Res.(*eth.BlockBodiesResponse).Unpack()
 	hashsets := packet.Meta.([][]common.Hash) // {txs hashes, uncle hashes, withdrawal hashes}
 
-	accepted, err := q.queue.DeliverBodies(peer.id, txs, hashsets[0], uncles, hashsets[1], withdrawals, hashsets[2])
+	accepted, err := q.queue.DeliverBodies(peer.id, txs, hashsets[0], uncles, hashsets[1], withdrawals, hashsets[2], sidecars)
 	switch {
 	case err == nil && len(txs) == 0:
 		peer.log.Trace("Requested bodies delivered")

eth/downloader/queue.go

@@ -71,6 +71,7 @@ type fetchResult struct {
 	Transactions types.Transactions
 	Receipts     types.Receipts
 	Withdrawals  types.Withdrawals
+	Sidecars     types.BlobTxSidecars
 }
 
 func newFetchResult(header *types.Header, fastSync bool, pid string) *fetchResult {
@@ -776,7 +777,7 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, hashes []comm
 // also wakes any threads waiting for data delivery.
 func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, txListHashes []common.Hash,
 	uncleLists [][]*types.Header, uncleListHashes []common.Hash,
-	withdrawalLists [][]*types.Withdrawal, withdrawalListHashes []common.Hash) (int, error) {
+	withdrawalLists [][]*types.Withdrawal, withdrawalListHashes []common.Hash, sidecars [][]*types.BlobTxSidecar) (int, error) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
@@ -838,6 +839,7 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, txListH
 		result.Transactions = txLists[index]
 		result.Uncles = uncleLists[index]
 		result.Withdrawals = withdrawalLists[index]
+		result.Sidecars = sidecars[index]
 		result.SetBodyDone()
 	}
 	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool,

eth/downloader/queue_test.go

@@ -341,7 +341,7 @@ func XTestDelivery(t *testing.T) {
 					uncleHashes[i] = types.CalcUncleHash(uncles)
 				}
 				time.Sleep(100 * time.Millisecond)
-				_, err := q.DeliverBodies(peer.id, txset, txsHashes, uncleset, uncleHashes, nil, nil)
+				_, err := q.DeliverBodies(peer.id, txset, txsHashes, uncleset, uncleHashes, nil, nil, nil)
 				if err != nil {
 					fmt.Printf("delivered %d bodies %v\n", len(txset), err)
 				}

eth/fetcher/block_fetcher.go

@@ -576,7 +576,8 @@ func (f *BlockFetcher) loop() {
 					case res := <-resCh:
 						res.Done <- nil
 						// Ignoring withdrawals here, since the block fetcher is not used post-merge.
-						txs, uncles, _ := res.Res.(*eth.BlockBodiesResponse).Unpack()
+						// todo 4844 is it ok to ignore sidecars here too?
+						txs, uncles, _, _ := res.Res.(*eth.BlockBodiesResponse).Unpack()
 						f.FilterBodies(peer, txs, uncles, time.Now())
 
 					case <-timeout.C:

eth/handler.go

@@ -794,6 +794,7 @@ func (h *handler) BroadcastBlock(block *types.Block, propagate bool) {
 		} else {
 			transfer = peers[:int(math.Sqrt(float64(len(peers))))]
 		}
+
 		for _, peer := range transfer {
 			peer.AsyncSendNewBlock(block, td)
 		}

eth/handler_eth.go

@@ -65,7 +65,7 @@ func (h *ethHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
 		return h.handleBlockAnnounces(peer, hashes, numbers)
 
 	case *eth.NewBlockPacket:
-		return h.handleBlockBroadcast(peer, packet.Block, packet.TD)
+		return h.handleBlockBroadcast(peer, packet)
 
 	case *eth.NewPooledTransactionHashesPacket67:
 		return h.txFetcher.Notify(peer.ID(), nil, nil, *packet)
@@ -118,13 +118,20 @@ func (h *ethHandler) handleBlockAnnounces(peer *eth.Peer, hashes []common.Hash,
 
 // handleBlockBroadcast is invoked from a peer's message handler when it transmits a
 // block broadcast for the local node to process.
-func (h *ethHandler) handleBlockBroadcast(peer *eth.Peer, block *types.Block, td *big.Int) error {
+func (h *ethHandler) handleBlockBroadcast(peer *eth.Peer, packet *eth.NewBlockPacket) error {
 	// Drop all incoming block announces from the p2p network if
 	// the chain already entered the pos stage and disconnect the
 	// remote peer.
 	if h.merger.PoSFinalized() {
 		return errors.New("disallowed block broadcast")
 	}
+	block := packet.Block
+	td := packet.TD
+	sidecars := packet.Sidecars
+	if sidecars != nil {
+		block = block.WithBlobs(sidecars)
+	}
+
 	// Schedule the block for import
 	h.blockFetcher.Enqueue(peer.ID(), block)
 

eth/protocols/eth/broadcast.go

@@ -33,8 +33,9 @@ const (
 // blockPropagation is a block propagation event, waiting for its turn in the
 // broadcast queue.
 type blockPropagation struct {
-	block *types.Block
-	td    *big.Int
+	block    *types.Block
+	td       *big.Int
+	sidecars types.BlobTxSidecars `rlp:"optional"`
 }
 
 // broadcastBlocks is a write loop that multiplexes blocks and block announcements
@@ -47,7 +48,11 @@ func (p *Peer) broadcastBlocks() {
 			if err := p.SendNewBlock(prop.block, prop.td); err != nil {
 				return
 			}
-			p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td)
+			if len(prop.sidecars) > 0 {
+				p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td, "sidecars", prop.sidecars.Len())
+			} else {
+				p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td)
+			}
 
 		case block := <-p.queuedBlockAnns:
 			if err := p.SendNewBlockHashes([]common.Hash{block.Hash()}, []uint64{block.NumberU64()}); err != nil {

eth/protocols/eth/handler_test.go

@@ -17,6 +17,8 @@
 package eth
 
 import (
+	rand2 "crypto/rand"
+	"io"
 	"math"
 	"math/big"
 	"math/rand"
@@ -33,10 +35,14 @@
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
+	"github.com/ethereum/go-ethereum/eth/protocols/bsc"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/params"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/holiman/uint256"
 )
 
 var (
@@ -146,7 +152,7 @@
 	panic("data processing tests should be done in the handler package")
 }
 func (b *testBackend) Handle(*Peer, Packet) error {
-	panic("data processing tests should be done in the handler package")
+	return nil
 }
 
 // Tests that block headers can be retrieved from a remote chain based on user queries.
@@ -502,3 +508,157 @@
 		t.Errorf("receipts mismatch: %v", err)
 	}
 }
+
+func TestHandleNewBlock(t *testing.T) {
+	t.Parallel()
+
+	gen := func(n int, g *core.BlockGen) {
+		if n%2 == 0 {
+			w := &types.Withdrawal{
+				Address: common.Address{0xaa},
+				Amount:  42,
+			}
+			g.AddWithdrawal(w)
+		}
+	}
+
+	backend := newTestBackendWithGenerator(maxBodiesServe+15, true, gen)
+	defer backend.close()
+
+	peer, _ := newTestPeer("peer", ETH68, backend)
+	defer peer.close()
+
+	v := new(uint32)
+	*v = 1
+	genBlobs := makeBlkBlobs(1, 2)
+	tx1 := types.NewTx(&types.BlobTx{
+		ChainID:    new(uint256.Int).SetUint64(1),
+		GasTipCap:  new(uint256.Int),
+		GasFeeCap:  new(uint256.Int),
+		Gas:        0,
+		Value:      new(uint256.Int),
+		Data:       nil,
+		BlobFeeCap: new(uint256.Int),
+		BlobHashes: []common.Hash{common.HexToHash("0x34ec6e64f9cda8fe0451a391e4798085a3ef51a65ed1bfb016e34fc1a2028f8f"), common.HexToHash("0xb9a412e875f29fac436acde234f954e91173c4cf79814f6dcf630d8a6345747f")},
+		Sidecar:    genBlobs[0],
+		V:          new(uint256.Int),
+		R:          new(uint256.Int),
+		S:          new(uint256.Int),
+	})
+	sidecars := types.BlobTxSidecars{tx1.BlobTxSidecar()}
+	block := types.NewBlockWithHeader(&types.Header{
+		Number:      big.NewInt(0),
+		Extra:       []byte("test block"),
+		UncleHash:   types.EmptyUncleHash,
+		TxHash:      types.EmptyTxsHash,
+		ReceiptHash: types.EmptyReceiptsHash,
+	})
+	dataNil := NewBlockPacket{
+		Block:    block,
+		TD:       big.NewInt(1),
+		Sidecars: nil,
+	}
+	dataNonNil := NewBlockPacket{
+		Block:    block,
+		TD:       big.NewInt(1),
+		Sidecars: sidecars,
+	}
+	sizeNonNil, rNonNil, _ := rlp.EncodeToReader(dataNonNil)
+	sizeNil, rNil, _ := rlp.EncodeToReader(dataNil)
+
+	// Define the test cases
+	testCases := []struct {
+		name string
+		msg  p2p.Msg
+		err  error
+	}{
+		{
+			name: "Valid block",
+			msg: p2p.Msg{
+				Code:    1,
+				Size:    uint32(sizeNonNil),
+				Payload: rNonNil,
+			},
+			err: nil,
+		},
+		{
+			name: "Nil sidecars",
+			msg: p2p.Msg{
+				Code:    2,
+				Size:    uint32(sizeNil),
+				Payload: rNil,
+			},
+			err: nil,
+		},
+	}
+
+	protos := []p2p.Protocol{
+		{
+			Name:    "eth",
+			Version: ETH67,
+		},
+		{
+			Name:    "eth",
+			Version: ETH68,
+		},
+		{
+			Name:    "bsc",
+			Version: bsc.Bsc1,
+		},
+	}
+	caps := []p2p.Cap{
+		{
+			Name:    "eth",
+			Version: ETH67,
+		},
+		{
+			Name:    "eth",
+			Version: ETH68,
+		},
+		{
+			Name:    "bsc",
+			Version: bsc.Bsc1,
+		},
+	}
+	// Create a source handler to send messages through and a sink peer to receive them
+	p2pEthSrc, p2pEthSink := p2p.MsgPipe()
+	defer p2pEthSrc.Close()
+	defer p2pEthSink.Close()
+
+	localEth := NewPeer(ETH68, p2p.NewPeerWithProtocols(enode.ID{1}, protos, "", caps), p2pEthSrc, nil)
+
+	// Run the tests
+	for _, tc := range testCases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			err := handleNewBlock(backend, tc.msg, localEth)
+			if err != tc.err {
+				t.Errorf("expected error %v, got %v", tc.err, err)
+			}
+		})
+	}
+
+}
+
+func makeBlkBlobs(n, nPerTx int) types.BlobTxSidecars {
+	if n <= 0 {
+		return nil
+	}
+	ret := make(types.BlobTxSidecars, n)
+	for i := 0; i < n; i++ {
+		blobs := make([]kzg4844.Blob, nPerTx)
+		commitments := make([]kzg4844.Commitment, nPerTx)
+		proofs := make([]kzg4844.Proof, nPerTx)
+		for i := 0; i < nPerTx; i++ {
+			io.ReadFull(rand2.Reader, blobs[i][:])
+			commitments[i], _ = kzg4844.BlobToCommitment(blobs[i])
+			proofs[i], _ = kzg4844.ComputeBlobProof(blobs[i], commitments[i])
+		}
+		ret[i] = &types.BlobTxSidecar{
+			Blobs:       blobs,
+			Commitments: commitments,
+			Proofs:      proofs,
+		}
+	}
+	return ret
+}

eth/protocols/eth/handlers.go

@@ -224,10 +224,24 @@ func ServiceGetBlockBodiesQuery(chain *core.BlockChain, query GetBlockBodiesRequ
 			lookups >= 2*maxBodiesServe {
 			break
 		}
-		if data := chain.GetBodyRLP(hash); len(data) != 0 {
-			bodies = append(bodies, data)
-			bytes += len(data)
+		body := chain.GetBody(hash)
+		if body == nil {
+			continue
+		}
+		blobs := chain.GetBlobsByHash(hash)
+		bodyWithBlobs := &BlockBody{
+			Transactions: body.Transactions,
+			Uncles:       body.Uncles,
+			Withdrawals:  body.Withdrawals,
+			Sidecars:     blobs,
+		}
+		enc, err := rlp.EncodeToBytes(bodyWithBlobs)
+		if err != nil {
+			log.Error("block body encode err", "hash", hash, "err", err)
+			continue
 		}
+		bodies = append(bodies, enc)
+		bytes += len(enc)
 	}
 	return bodies
 }
@@ -293,9 +307,12 @@ func handleNewBlock(backend Backend, msg Decoder, peer *Peer) error {
 	if err := msg.Decode(ann); err != nil {
 		return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
 	}
+
+	// Now that we have our packet, perform operations using the interface methods
 	if err := ann.sanityCheck(); err != nil {
 		return err
 	}
+
 	if hash := types.CalcUncleHash(ann.Block.Uncles()); hash != ann.Block.UncleHash() {
 		log.Warn("Propagated block has invalid uncles", "have", hash, "exp", ann.Block.UncleHash())
 		return nil // TODO(karalabe): return error eventually, but wait a few releases