fix: change the decode type

eth/downloader/downloader.go

@@ -1049,7 +1049,7 @@ func (d *Downloader) commitSnapSyncData(results []*fetchResult, stateSync *state
 	receipts := make([]rlp.RawValue, len(results))
 	for i, result := range results {
 		blocks[i] = types.NewBlockWithHeader(result.Header).WithBody(result.body())
-		receipts[i] = result.Receipts
+		receipts[i] = types.EncodeBlockReceiptLists([]types.Receipts{result.Receipts})[0]
 	}
 	if index, err := d.blockchain.InsertReceiptChain(blocks, receipts, d.ancientLimit); err != nil {
 		log.Debug("Downloaded item processing failed", "number", results[index].Header.Number, "hash", results[index].Header.Hash(), "err", err)
@@ -1063,7 +1063,7 @@ func (d *Downloader) commitPivotBlock(result *fetchResult) error {
 	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
-	if _, err := d.blockchain.InsertReceiptChain([]*types.Block{block}, []rlp.RawValue{result.Receipts}, d.ancientLimit); err != nil {
+	if _, err := d.blockchain.InsertReceiptChain([]*types.Block{block}, types.EncodeBlockReceiptLists([]types.Receipts{result.Receipts}), d.ancientLimit); err != nil {
 		return err
 	}
 	if err := d.blockchain.SnapSyncCommitHead(block.Hash()); err != nil {

eth/downloader/fetchers_concurrent_receipts.go

@@ -95,12 +95,15 @@ func (q *receiptQueue) deliver(peer *peerConnection, packet *eth.Response) (int,
 
 	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")
+		var rcStorage []*types.ReceiptForStorage
+		if err := rlp.DecodeBytes(rc, &rcStorage); err != nil {
+			peer.log.Error("Failed to decode retreived receipts", "err", err)
 			return 0, err
 		}
-		receipts[i] = r
+		receipts[i] = make([]*types.Receipt, len(rcStorage))
+		for j := range rcStorage {
+			receipts[i][j] = (*types.Receipt)(rcStorage[j])
+		}
 	}
 
 	accepted, err := q.queue.DeliverReceipts(peer.id, receipts, hashes, false)

eth/downloader/queue.go

@@ -70,7 +70,7 @@ type fetchResult struct {
 	Header       *types.Header
 	Uncles       []*types.Header
 	Transactions types.Transactions
-	Receipts     [][]*types.Receipt
+	Receipts     types.Receipts
 	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 = [][]*types.Receipt{}
+			item.Receipts = []*types.Receipt{}
 		} else {
 			item.pending.Store(item.pending.Load() | (1 << receiptType))
 		}
@@ -647,11 +647,11 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt, recei
 		return nil
 	}
 	reconstruct := func(index int, result *fetchResult) int {
-		result.Receipts = append(result.Receipts, receiptList[index])
-		if index == len(receiptList) && lastBlockIncomplete {
+		result.Receipts = append(result.Receipts, receiptList[index]...)
+		if lastBlockIncomplete && index == len(receiptList) {
 			// 1. Verify the total number of tx delivered
 			if uint64(len(receiptList[index])) > result.Header.GasUsed/21_000 {
-				result.Receipts = [][]*types.Receipt{}
+				result.Receipts = []*types.Receipt{}
 				return 0
 			}
 			// 2. Verify the size of each receipt against the gas limit of the corresponding transaction
@@ -661,14 +661,14 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt, recei
 					logSize += uint64(len(log.Data))
 				}
 				if logSize > params.MaxTxGas/params.LogDataGas {
-					result.Receipts = [][]*types.Receipt{}
+					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{}
+				result.Receipts = []*types.Receipt{}
 				return 0
 			}
 			return len(result.Receipts)
@@ -730,13 +730,9 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 	}
 
 	for _, header := range request.Headers[:i] {
+		resume := -1
 		if res, stale, err := q.resultCache.GetDeliverySlot(header.Number.Uint64()); err == nil && !stale {
-			resume := reconstruct(accepted, res)
-			if resume >= 0 {
-				// TODO: add receipt index in TaskPool
-				taskPool[header.Hash()] = header
-				taskQueue.Push(header, -int64(header.Number.Uint64()))
-			}
+			resume = reconstruct(accepted, res)
 		} 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
@@ -746,6 +742,11 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 		}
 		// Clean up a successful fetch
 		delete(taskPool, hashes[accepted])
+		if resume >= 0 {
+			// TODO: add receipt index in TaskPool
+			taskPool[header.Hash()] = header
+			taskQueue.Push(header, -int64(header.Number.Uint64()))
+		}
 		accepted++
 	}
 	resDropMeter.Mark(int64(results - accepted))

eth/downloader/queue_test.go

@@ -367,7 +367,11 @@ func XTestDelivery(t *testing.T) {
 				for i, receipt := range rcs {
 					hashes[i] = types.DeriveSha(receipt, hasher)
 				}
-				_, err := q.DeliverReceipts(peer.id, types.EncodeBlockReceiptLists(rcs), hashes)
+				receiptSets := make([][]*types.Receipt, len(rcs))
+				for i := range rcs {
+					receiptSets[i] = rcs[i]
+				}
+				_, err := q.DeliverReceipts(peer.id, receiptSets, hashes, false)
 				if err != nil {
 					fmt.Printf("delivered %d receipts %v\n", len(rcs), err)
 				}