add: add GetReceiptsPacket69

cmd/devp2p/internal/ethtest/suite.go

@@ -434,7 +434,7 @@ func (s *Suite) TestGetReceipts(t *utesting.T) {
 	}
 
 	// Create block bodies request.
-	req := &eth.GetReceiptsPacket{
+	req := &eth.GetReceiptsPacket69{
 		RequestId:          66,
 		GetReceiptsRequest: (eth.GetReceiptsRequest)(hashes),
 	}

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))

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

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.

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
@@ -146,6 +144,7 @@ type queue struct {
 	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)
@@ -165,6 +164,7 @@ func newQueue(blockCacheLimit int, thresholdInitialSize int) *queue {
 		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,
 	}
@@ -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,24 +468,31 @@ 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 {
+		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
 		q.active.Signal()
@@ -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++
 	}

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
 }

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")
 }
 

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,
 	})

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
 	}

eth/protocols/eth/peer.go

@@ -319,20 +319,34 @@ 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{
+	var req *Request
+	if p.version < ETH70 {
+		req = &Request{
 			id:   id,
 			sink: sink,
 			code: GetReceiptsMsg,
 			want: ReceiptsMsg,
-		data: &GetReceiptsPacket{
+			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
 	}

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
 

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"),
 		},
 		{