diff --git a/cmd/devp2p/internal/ethtest/suite.go b/cmd/devp2p/internal/ethtest/suite.go index c23360bf82..956a12bd5b 100644 --- a/cmd/devp2p/internal/ethtest/suite.go +++ b/cmd/devp2p/internal/ethtest/suite.go @@ -434,7 +434,7 @@ func (s *Suite) TestGetReceipts(t *utesting.T) { } // Create block bodies request. - req := ð.GetReceiptsPacket{ + req := ð.GetReceiptsPacket69{ RequestId: 66, GetReceiptsRequest: (eth.GetReceiptsRequest)(hashes), } diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index c1a31d6e1c..98a9000050 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -253,7 +253,7 @@ func (dlp *downloadTesterPeer) RequestBodies(hashes []common.Hash, sink chan *et // RequestReceipts constructs a getReceipts method associated with a particular // peer in the download tester. The returned function can be used to retrieve // batches of block receipts from the particularly requested peer. -func (dlp *downloadTesterPeer) RequestReceipts(hashes []common.Hash, sink chan *eth.Response) (*eth.Request, error) { +func (dlp *downloadTesterPeer) RequestReceipts(hashes []common.Hash, _ uint64, sink chan *eth.Response) (*eth.Request, error) { blobs := eth.ServiceGetReceiptsQuery68(dlp.chain, hashes) receipts := make([]types.Receipts, len(blobs)) diff --git a/eth/downloader/fetchers_concurrent_receipts.go b/eth/downloader/fetchers_concurrent_receipts.go index 1eaffd7cea..95710d4a06 100644 --- a/eth/downloader/fetchers_concurrent_receipts.go +++ b/eth/downloader/fetchers_concurrent_receipts.go @@ -84,7 +84,7 @@ func (q *receiptQueue) request(peer *peerConnection, req *fetchRequest, resCh ch for _, header := range req.Headers { hashes = append(hashes, header.Hash()) } - return peer.peer.RequestReceipts(hashes, resCh) + return peer.peer.RequestReceipts(hashes, req.From, resCh) } // deliver is responsible for taking a generic response packet from the concurrent diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 0848e92a26..b21e92593e 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -60,7 +60,7 @@ type Peer interface { RequestHeadersByNumber(uint64, int, int, bool, chan *eth.Response) (*eth.Request, error) RequestBodies([]common.Hash, chan *eth.Response) (*eth.Request, error) - RequestReceipts([]common.Hash, chan *eth.Response) (*eth.Request, error) + RequestReceipts([]common.Hash, uint64, chan *eth.Response) (*eth.Request, error) } // newPeerConnection creates a new downloader peer. diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 35d418cb73..63c7436d27 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -57,18 +57,16 @@ var ( // fetchRequest is a currently running data retrieval operation. type fetchRequest struct { Peer *peerConnection // Peer to which the request was sent - From uint64 // Requested chain element index (used for skeleton fills only) + From uint64 // Requested chain element index (used for skeleton fills and receipt fetching only) Headers []*types.Header // Requested headers, sorted by request order Time time.Time // Time when the request was made - - FromIndex uint64 // TODO: Can we reuse `From` field here? } // fetchResult is a struct collecting partial results from data fetchers until // 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 + receiptSize uint64 // Size of log collected Header *types.Header Uncles []*types.Header @@ -142,10 +140,11 @@ type queue struct { blockPendPool map[string]*fetchRequest // Currently pending block (body) retrieval operations blockWakeCh chan bool // Channel to notify the block fetcher of new tasks - receiptTaskPool map[common.Hash]*types.Header // Pending receipt retrieval tasks, mapping hashes to headers - receiptTaskQueue *prque.Prque[int64, *types.Header] // Priority queue of the headers to fetch the receipts for - receiptPendPool map[string]*fetchRequest // Currently pending receipt retrieval operations - receiptWakeCh chan bool // Channel to notify when receipt fetcher of new tasks + receiptTaskPool map[common.Hash]*types.Header // Pending receipt retrieval tasks, mapping hashes to headers + receiptTaskQueue *prque.Prque[int64, *types.Header] // Priority queue of the headers to fetch the receipts for + receiptPendPool map[string]*fetchRequest // Currently pending receipt retrieval operations + receiptWakeCh chan bool // Channel to notify when receipt fetcher of new tasks + partialReceiptQueue *prque.Prque[int64, *types.Header] resultCache *resultStore // Downloaded but not yet delivered fetch results resultSize common.StorageSize // Approximate size of a block (exponential moving average) @@ -161,12 +160,13 @@ type queue struct { func newQueue(blockCacheLimit int, thresholdInitialSize int) *queue { lock := new(sync.RWMutex) q := &queue{ - blockTaskQueue: prque.New[int64, *types.Header](nil), - blockWakeCh: make(chan bool, 1), - receiptTaskQueue: prque.New[int64, *types.Header](nil), - receiptWakeCh: make(chan bool, 1), - active: sync.NewCond(lock), - lock: lock, + blockTaskQueue: prque.New[int64, *types.Header](nil), + blockWakeCh: make(chan bool, 1), + receiptTaskQueue: prque.New[int64, *types.Header](nil), + receiptWakeCh: make(chan bool, 1), + partialReceiptQueue: prque.New[int64, *types.Header](nil), + active: sync.NewCond(lock), + lock: lock, } q.Reset(blockCacheLimit, thresholdInitialSize) return q @@ -188,6 +188,7 @@ func (q *queue) Reset(blockCacheLimit int, thresholdInitialSize int) { q.receiptTaskPool = make(map[common.Hash]*types.Header) q.receiptTaskQueue.Reset() q.receiptPendPool = make(map[string]*fetchRequest) + q.partialReceiptQueue.Reset() q.resultCache = newResultStore(blockCacheLimit) q.resultCache.SetThrottleThreshold(uint64(thresholdInitialSize)) @@ -215,7 +216,7 @@ func (q *queue) PendingReceipts() int { q.lock.Lock() defer q.lock.Unlock() - return q.receiptTaskQueue.Size() + return q.receiptTaskQueue.Size() + q.partialReceiptQueue.Size() } // InFlightBlocks retrieves whether there are block fetch requests currently in @@ -419,12 +420,22 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common // Retrieve a batch of tasks, skipping previously failed ones send := make([]*types.Header, 0, count) skip := make([]*types.Header, 0) + skipPartial := make(map[*types.Header]bool) // Track which headers came from partialReceiptQueue progress := false throttled := false for proc := 0; len(send) < count && !taskQueue.Empty(); proc++ { + var selectedQueue *prque.Prque[int64, *types.Header] + partial := false + if kind == receiptType && p.version >= 70 && len(send) == 0 && !q.partialReceiptQueue.Empty() { + selectedQueue = q.partialReceiptQueue + partial = true + } else { + selectedQueue = taskQueue + } + // the task queue will pop items in order, so the highest prio block // is also the lowest block number. - header, _ := taskQueue.Peek() + header, _ := selectedQueue.Peek() // we can ask the resultcache if this header is within the // "prioritized" segment of blocks. If it is not, we need to throttle @@ -433,7 +444,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common if stale { // Don't put back in the task queue, this item has already been // delivered upstream - taskQueue.PopItem() + selectedQueue.PopItem() progress = true delete(taskPool, header.Hash()) proc = proc - 1 @@ -457,23 +468,30 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common if item.Done(kind) { // If it's a noop, we can skip this task delete(taskPool, header.Hash()) - taskQueue.PopItem() + selectedQueue.PopItem() proc = proc - 1 progress = true continue } // Remove it from the task queue - taskQueue.PopItem() + selectedQueue.PopItem() // Otherwise unless the peer is known not to have the data, add to the retrieve list if p.Lacks(header.Hash()) { skip = append(skip, header) + if partial { + skipPartial[header] = true + } } else { send = append(send, header) } } - // Merge all the skipped headers back + // Merge all the skipped headers back to their original queues for _, header := range skip { - taskQueue.Push(header, -int64(header.Number.Uint64())) + if skipPartial[header] { + q.partialReceiptQueue.Push(header, -int64(header.Number.Uint64())) + } else { + taskQueue.Push(header, -int64(header.Number.Uint64())) + } } if q.resultCache.HasCompletedItems() { // Wake Results, resultCache was modified @@ -488,6 +506,15 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common Headers: send, Time: time.Now(), } + + // If this is a receipt request and the first block has a partial delivery, + // set the From field to resume from the last index + if kind == receiptType { + if result, _, _, _, err := q.resultCache.getFetchResult(send[0].Number.Uint64()); err == nil && result != nil { + request.From = uint64(len(result.Receipts)) + } + } + pendPool[p.id] = request return request, progress, throttled } @@ -632,7 +659,7 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, txListH result.SetBodyDone() return -1 } - return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, + return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, nil, bodyReqTimer, bodyInMeter, bodyDropMeter, len(txLists), validate, reconstruct) } @@ -709,7 +736,7 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt, recei result.SetReceiptsDone() return -1 } - return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, + return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.partialReceiptQueue, receiptReqTimer, receiptInMeter, receiptDropMeter, len(receiptList), validate, reconstruct) } @@ -720,6 +747,7 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt, recei // to access the queue, so they already need a lock anyway. func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque[int64, *types.Header], pendPool map[string]*fetchRequest, + partialQueue *prque.Prque[int64, *types.Header], reqTimer *metrics.Timer, resInMeter, resDropMeter *metrics.Meter, resultCount int, validate func(index int, header *types.Header, result *fetchResult) error, reconstruct func(index int, result *fetchResult) int) (int, error) { @@ -779,10 +807,9 @@ 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 + if partialQueue != nil && resume >= 0 { taskPool[header.Hash()] = header - taskQueue.Push(header, -int64(header.Number.Uint64())) + partialQueue.Push(header, -int64(header.Number.Uint64())) } accepted++ } diff --git a/eth/downloader/queue_test.go b/eth/downloader/queue_test.go index 0987bf9ef6..c601a9b7a5 100644 --- a/eth/downloader/queue_test.go +++ b/eth/downloader/queue_test.go @@ -100,6 +100,7 @@ func dummyPeer(id string) *peerConnection { p := &peerConnection{ id: id, lacking: make(map[common.Hash]struct{}), + version: 70, } return p } diff --git a/eth/downloader/skeleton_test.go b/eth/downloader/skeleton_test.go index 4aa97cf1f7..3f8b43f6ca 100644 --- a/eth/downloader/skeleton_test.go +++ b/eth/downloader/skeleton_test.go @@ -201,7 +201,7 @@ func (p *skeletonTestPeer) RequestBodies([]common.Hash, chan *eth.Response) (*et panic("skeleton sync must not request block bodies") } -func (p *skeletonTestPeer) RequestReceipts([]common.Hash, chan *eth.Response) (*eth.Request, error) { +func (p *skeletonTestPeer) RequestReceipts([]common.Hash, uint64, chan *eth.Response) (*eth.Request, error) { panic("skeleton sync must not request receipts") } diff --git a/eth/protocols/eth/handler_test.go b/eth/protocols/eth/handler_test.go index 65c491f815..4143591fcd 100644 --- a/eth/protocols/eth/handler_test.go +++ b/eth/protocols/eth/handler_test.go @@ -538,7 +538,7 @@ func testGetBlockReceipts(t *testing.T, protocol uint) { } // Send the hash request and verify the response - p2p.Send(peer.app, GetReceiptsMsg, &GetReceiptsPacket{ + p2p.Send(peer.app, GetReceiptsMsg, &GetReceiptsPacket69{ RequestId: 123, GetReceiptsRequest: hashes, }) diff --git a/eth/protocols/eth/handlers.go b/eth/protocols/eth/handlers.go index aad3353d88..808e15372c 100644 --- a/eth/protocols/eth/handlers.go +++ b/eth/protocols/eth/handlers.go @@ -250,7 +250,7 @@ func ServiceGetBlockBodiesQuery(chain *core.BlockChain, query GetBlockBodiesRequ func handleGetReceipts68(backend Backend, msg Decoder, peer *Peer) error { // Decode the block receipts retrieval message - var query GetReceiptsPacket + var query GetReceiptsPacket69 if err := msg.Decode(&query); err != nil { return err } @@ -260,7 +260,7 @@ func handleGetReceipts68(backend Backend, msg Decoder, peer *Peer) error { func handleGetReceipts69(backend Backend, msg Decoder, peer *Peer) error { // Decode the block receipts retrieval message - var query GetReceiptsPacket + var query GetReceiptsPacket69 if err := msg.Decode(&query); err != nil { return err } diff --git a/eth/protocols/eth/peer.go b/eth/protocols/eth/peer.go index 40c54a3570..34b427fec1 100644 --- a/eth/protocols/eth/peer.go +++ b/eth/protocols/eth/peer.go @@ -319,19 +319,33 @@ func (p *Peer) RequestBodies(hashes []common.Hash, sink chan *Response) (*Reques } // RequestReceipts fetches a batch of transaction receipts from a remote node. -func (p *Peer) RequestReceipts(hashes []common.Hash, sink chan *Response) (*Request, error) { +func (p *Peer) RequestReceipts(hashes []common.Hash, firstBlockReceiptIndex uint64, sink chan *Response) (*Request, error) { p.Log().Debug("Fetching batch of receipts", "count", len(hashes)) id := rand.Uint64() - - req := &Request{ - id: id, - sink: sink, - code: GetReceiptsMsg, - want: ReceiptsMsg, - data: &GetReceiptsPacket{ - RequestId: id, - GetReceiptsRequest: hashes, - }, + var req *Request + if p.version < ETH70 { + req = &Request{ + id: id, + sink: sink, + code: GetReceiptsMsg, + want: ReceiptsMsg, + data: &GetReceiptsPacket69{ + RequestId: id, + GetReceiptsRequest: hashes, + }, + } + } else { + req = &Request{ + id: id, + sink: sink, + code: GetReceiptsMsg, + want: ReceiptsMsg, + data: &GetReceiptsPacket70{ + RequestId: id, + GetReceiptsRequest: hashes, + FirstBlockReceiptIndex: firstBlockReceiptIndex, + }, + } } if err := p.dispatchRequest(req); err != nil { return nil, err diff --git a/eth/protocols/eth/protocol.go b/eth/protocols/eth/protocol.go index 7c41e7a996..8ce09788ae 100644 --- a/eth/protocols/eth/protocol.go +++ b/eth/protocols/eth/protocol.go @@ -32,6 +32,7 @@ import ( const ( ETH68 = 68 ETH69 = 69 + ETH70 = 70 ) // ProtocolName is the official short name of the `eth` protocol used during @@ -259,11 +260,18 @@ func (p *BlockBodiesResponse) Unpack() ([][]*types.Transaction, [][]*types.Heade type GetReceiptsRequest []common.Hash // GetReceiptsPacket represents a block receipts query with request ID wrapping. -type GetReceiptsPacket struct { +type GetReceiptsPacket69 struct { RequestId uint64 GetReceiptsRequest } +// GetReceiptsPacket represents a block receipts query with request ID wrapping. +type GetReceiptsPacket70 struct { + RequestId uint64 + GetReceiptsRequest + FirstBlockReceiptIndex uint64 +} + // ReceiptsResponse is the network packet for block receipts distribution. type ReceiptsResponse []types.Receipts diff --git a/eth/protocols/eth/protocol_test.go b/eth/protocols/eth/protocol_test.go index 8a2559a6c5..3c6737cbb0 100644 --- a/eth/protocols/eth/protocol_test.go +++ b/eth/protocols/eth/protocol_test.go @@ -84,7 +84,7 @@ func TestEmptyMessages(t *testing.T) { BlockBodiesPacket{1111, nil}, BlockBodiesRLPPacket{1111, nil}, // Receipts - GetReceiptsPacket{1111, nil}, + GetReceiptsPacket69{1111, nil}, // Transactions GetPooledTransactionsPacket{1111, nil}, PooledTransactionsPacket{1111, nil}, @@ -97,7 +97,7 @@ func TestEmptyMessages(t *testing.T) { BlockBodiesPacket{1111, BlockBodiesResponse([]*BlockBody{})}, BlockBodiesRLPPacket{1111, BlockBodiesRLPResponse([]rlp.RawValue{})}, // Receipts - GetReceiptsPacket{1111, GetReceiptsRequest([]common.Hash{})}, + GetReceiptsPacket69{1111, GetReceiptsRequest([]common.Hash{})}, ReceiptsPacket[*ReceiptList68]{1111, []*ReceiptList68{}}, ReceiptsPacket[*ReceiptList69]{1111, []*ReceiptList69{}}, // Transactions @@ -227,7 +227,7 @@ func TestMessages(t *testing.T) { common.FromHex("f902dc820457f902d6f902d3f8d2f867088504a817c8088302e2489435353535353535353535353535353535353535358202008025a064b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c12a064b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c10f867098504a817c809830334509435353535353535353535353535353535353535358202d98025a052f8f61201b2b11a78d6e866abc9c3db2ae8631fa656bfe5cb53668255367afba052f8f61201b2b11a78d6e866abc9c3db2ae8631fa656bfe5cb53668255367afbf901fcf901f9a00000000000000000000000000000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000940000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000b90100000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000008208ae820d0582115c8215b3821a0a827788a00000000000000000000000000000000000000000000000000000000000000000880000000000000000"), }, { - GetReceiptsPacket{1111, GetReceiptsRequest(hashes)}, + GetReceiptsPacket69{1111, GetReceiptsRequest(hashes)}, common.FromHex("f847820457f842a000000000000000000000000000000000000000000000000000000000deadc0dea000000000000000000000000000000000000000000000000000000000feedbeef"), }, {