diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 09837a3045..f410fd066c 100644 --- a/eth/downloader/downloader.go +++ b/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 { diff --git a/eth/downloader/fetchers_concurrent_receipts.go b/eth/downloader/fetchers_concurrent_receipts.go index 8655f1c7ab..1eaffd7cea 100644 --- a/eth/downloader/fetchers_concurrent_receipts.go +++ b/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) diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 3f240d4dcd..da856a710c 100644 --- a/eth/downloader/queue.go +++ b/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)) diff --git a/eth/downloader/queue_test.go b/eth/downloader/queue_test.go index 120e3f9d2d..b1bc2466ca 100644 --- a/eth/downloader/queue_test.go +++ b/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) }