add: add: validate and buffer partial receipts within DeliverReceipts()

eth/downloader/downloader.go

@@ -130,6 +130,7 @@ type Downloader struct {
 
 	// Channels
 	headerProcCh chan *headerTask // Channel to feed the header processor new tasks
+	bodyProcCh   chan *headerTask // Channle to feed the body processor new tasks
 
 	// Skeleton sync
 	skeleton *skeleton // Header skeleton to backfill the chain with (eth2 mode)
@@ -233,6 +234,7 @@ func New(stateDb ethdb.Database, mux *event.TypeMux, chain BlockChain, dropPeer
 		chainCutoffHash:   cutoffHash,
 		dropPeer:          dropPeer,
 		headerProcCh:      make(chan *headerTask, 1),
+		bodyProcCh:        make(chan *headerTask, 1), // todo Is it enough to set this buffer size as 1
 		quitCh:            make(chan struct{}),
 		SnapSyncer:        snap.NewSyncer(stateDb, chain.TrieDB().Scheme()),
 		stateSyncStart:    make(chan *stateSync),

eth/downloader/fetchers_concurrent_receipts.go

@@ -20,8 +20,10 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rlp"
 )
 
 // receiptQueue implements typedQueue and is a type adapter between the generic
@@ -88,10 +90,20 @@ func (q *receiptQueue) request(peer *peerConnection, req *fetchRequest, resCh ch
 // deliver is responsible for taking a generic response packet from the concurrent
 // fetcher, unpacking the receipt data and delivering it to the downloader's queue.
 func (q *receiptQueue) deliver(peer *peerConnection, packet *eth.Response) (int, error) {
-	receipts := *packet.Res.(*eth.ReceiptsRLPResponse)
+	receiptsRLP := *packet.Res.(*eth.ReceiptsRLPResponse)
 	hashes := packet.Meta.([]common.Hash) // {receipt hashes}
 
-	accepted, err := q.queue.DeliverReceipts(peer.id, receipts, hashes)
+	receipts := make([][]*types.Receipt, len(receiptsRLP))
+	for i, rc := range receiptsRLP {
+		var r []*types.Receipt
+		if err := rlp.DecodeBytes(rc, &r); err != nil {
+			peer.log.Debug("Failed to decode retreived receipts")
+			return 0, err
+		}
+		receipts[i] = r
+	}
+
+	accepted, err := q.queue.DeliverReceipts(peer.id, receipts, hashes, false)
 	switch {
 	case err == nil && len(receipts) == 0:
 		peer.log.Trace("Requested receipts delivered")

eth/downloader/queue.go

@@ -34,7 +34,6 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/ethereum/go-ethereum/rlp"
 )
 
 const (
@@ -66,11 +65,12 @@ type fetchRequest struct {
 // all outstanding pieces complete and the result as a whole can be processed.
 type fetchResult struct {
 	pending     atomic.Int32 // Flag telling what deliveries are outstanding
+	receiptSize uint64
 
 	Header       *types.Header
 	Uncles       []*types.Header
 	Transactions types.Transactions
-	Receipts     rlp.RawValue
+	Receipts     [][]*types.Receipt
 	Withdrawals  types.Withdrawals
 }
 
@@ -86,7 +86,7 @@ func newFetchResult(header *types.Header, snapSync bool) *fetchResult {
 	if snapSync {
 		if header.EmptyReceipts() {
 			// Ensure the receipts list is valid even if it isn't actively fetched.
-			item.Receipts = rlp.EmptyList
+			item.Receipts = [][]*types.Receipt{}
 		} else {
 			item.pending.Store(item.pending.Load() | (1 << receiptType))
 		}
@@ -622,11 +622,12 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, txListH
 		return nil
 	}
 
-	reconstruct := func(index int, result *fetchResult) {
+	reconstruct := func(index int, result *fetchResult) int {
 		result.Transactions = txLists[index]
 		result.Uncles = uncleLists[index]
 		result.Withdrawals = withdrawalLists[index]
 		result.SetBodyDone()
+		return -1
 	}
 	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool,
 		bodyReqTimer, bodyInMeter, bodyDropMeter, len(txLists), validate, reconstruct)
@@ -635,7 +636,7 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, txListH
 // DeliverReceipts injects a receipt retrieval response into the results queue.
 // The method returns the number of transaction receipts accepted from the delivery
 // and also wakes any threads waiting for data delivery.
-func (q *queue) DeliverReceipts(id string, receiptList []rlp.RawValue, receiptListHashes []common.Hash) (int, error) {
+func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt, receiptListHashes []common.Hash, lastBlockIncomplete bool) (int, error) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
@@ -645,9 +646,36 @@ func (q *queue) DeliverReceipts(id string, receiptList []rlp.RawValue, receiptLi
 		}
 		return nil
 	}
-	reconstruct := func(index int, result *fetchResult) {
+	reconstruct := func(index int, result *fetchResult) int {
-		result.Receipts = receiptList[index]
+		result.Receipts = append(result.Receipts, receiptList[index])
+		if index == len(receiptList) && lastBlockIncomplete {
+			// 1. Verify the total number of tx delivered
+			if uint64(len(receiptList[index])) > result.Header.GasUsed/21_000 {
+				result.Receipts = [][]*types.Receipt{}
+				return 0
+			}
+			// 2. Verify the size of each receipt against the gas limit of the corresponding transaction
+			for _, rc := range receiptList[index] {
+				var logSize uint64
+				for _, log := range rc.Logs {
+					logSize += uint64(len(log.Data))
+				}
+				if logSize > params.MaxTxGas/params.LogDataGas {
+					result.Receipts = [][]*types.Receipt{}
+					return 0
+				}
+				result.receiptSize += logSize
+			}
+			// 3. Verify the total download receipt size is no longer than allowed by the block gas limit
+			if result.receiptSize > result.Header.GasLimit/params.LogDataGas {
+				result.Receipts = [][]*types.Receipt{}
+				return 0
+			}
+			return len(result.Receipts)
+		}
+		// receipt fetch is done
 		result.SetReceiptsDone()
+		return -1
 	}
 	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool,
 		receiptReqTimer, receiptInMeter, receiptDropMeter, len(receiptList), validate, reconstruct)
@@ -662,7 +690,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 	taskQueue *prque.Prque[int64, *types.Header], pendPool map[string]*fetchRequest,
 	reqTimer *metrics.Timer, resInMeter, resDropMeter *metrics.Meter,
 	results int, validate func(index int, header *types.Header) error,
-	reconstruct func(index int, result *fetchResult)) (int, error) {
+	reconstruct func(index int, result *fetchResult) int) (int, error) {
 	// Short circuit if the data was never requested
 	request := pendPool[id]
 	if request == nil {
@@ -703,7 +731,11 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 
 	for _, header := range request.Headers[:i] {
 		if res, stale, err := q.resultCache.GetDeliverySlot(header.Number.Uint64()); err == nil && !stale {
-			reconstruct(accepted, res)
+			resume := reconstruct(accepted, res)
+			if resume >= 0 {
+				q.receiptTaskPool[header.Hash()] = header
+				q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
+			}
 		} else {
 			// else: between here and above, some other peer filled this result,
 			// or it was indeed a no-op. This should not happen, but if it does it's