fix: remove partial sink

Author: healthykim

eth/downloader/fetchers_concurrent.go

@@ -307,15 +307,7 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error {
 			// reschedule the timeout timer.
 			index, live := ordering[res.Req]
 			if live {
-				req := timeouts.Remove(index)
-				delete(ordering, res.Req)
-
-				if res.Partial {
-					ttl := d.peers.rates.TargetTimeout()
-					ordering[req] = timeouts.Size()
-					timeouts.Push(req, -time.Now().Add(ttl).UnixNano())
-				}
-
+				timeouts.Remove(index)
 				if index == 0 {
 					if !timeout.Stop() {
 						<-timeout.C
@@ -325,17 +317,16 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error {
 						timeout.Reset(time.Until(time.Unix(0, -exp)))
 					}
 				}
+				delete(ordering, res.Req)
 			}
-			if !res.Partial {
-				// Delete the pending request (if it still exists) and mark the peer idle
-				delete(pending, res.Req.Peer)
-				delete(stales, res.Req.Peer)
+			// Delete the pending request (if it still exists) and mark the peer idle
+			delete(pending, res.Req.Peer)
+			delete(stales, res.Req.Peer)
 
-				res.Req.Close()
-			}
 			// Signal the dispatcher that the round trip is done. We'll drop the
 			// peer if the data turns out to be junk.
 			res.Done <- nil
+			res.Req.Close()
 
 			// If the peer was previously banned and failed to deliver its pack
 			// in a reasonable time frame, ignore its message.

eth/downloader/fetchers_concurrent_receipts.go

@@ -91,7 +91,7 @@ func (q *receiptQueue) deliver(peer *peerConnection, packet *eth.Response) (int,
 	receipts := *packet.Res.(*eth.ReceiptsRLPResponse)
 	hashes := packet.Meta.([]common.Hash) // {receipt hashes}
 
-	accepted, err := q.queue.DeliverReceipts(peer.id, receipts, hashes, packet.Partial, packet.From)
+	accepted, err := q.queue.DeliverReceipts(peer.id, receipts, hashes)
 	switch {
 	case err == nil && len(receipts) == 0:
 		peer.log.Trace("Requested receipts delivered")

eth/downloader/queue.go

@@ -629,13 +629,13 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, txListH
 		result.SetBodyDone()
 	}
 	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool,
-		bodyReqTimer, bodyInMeter, bodyDropMeter, len(txLists), validate, reconstruct, false, 0)
+		bodyReqTimer, bodyInMeter, bodyDropMeter, len(txLists), validate, reconstruct)
 }
 
 // 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, incomplete bool, from int) (int, error) {
+func (q *queue) DeliverReceipts(id string, receiptList []rlp.RawValue, receiptListHashes []common.Hash) (int, error) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
@@ -650,7 +650,7 @@ func (q *queue) DeliverReceipts(id string, receiptList []rlp.RawValue, receiptLi
 		result.SetReceiptsDone()
 	}
 	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool,
-		receiptReqTimer, receiptInMeter, receiptDropMeter, len(receiptList), validate, reconstruct, incomplete, from)
+		receiptReqTimer, receiptInMeter, receiptDropMeter, len(receiptList), validate, reconstruct)
 }
 
 // deliver injects a data retrieval response into the results queue.
@@ -662,16 +662,14 @@ 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), incomplete bool, from int) (int, error) {
+	reconstruct func(index int, result *fetchResult)) (int, error) {
 	// Short circuit if the data was never requested
 	request := pendPool[id]
 	if request == nil {
 		resDropMeter.Mark(int64(results))
 		return 0, errNoFetchesPending
 	}
-	if !incomplete {
-		delete(pendPool, id)
-	}
+	delete(pendPool, id)
 
 	reqTimer.UpdateSince(request.Time)
 	resInMeter.Mark(int64(results))
@@ -689,7 +687,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 		i        int
 		hashes   []common.Hash
 	)
-	for _, header := range request.Headers[from:] {
+	for _, header := range request.Headers {
 		// Short circuit assembly if no more fetch results are found
 		if i >= results {
 			break
@@ -703,7 +701,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 		i++
 	}
 
-	for _, header := range request.Headers[from : from+i] {
+	for _, header := range request.Headers[:i] {
 		if res, stale, err := q.resultCache.GetDeliverySlot(header.Number.Uint64()); err == nil && !stale {
 			reconstruct(accepted, res)
 		} else {
@@ -720,14 +718,8 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 	resDropMeter.Mark(int64(results - accepted))
 
 	// Return all failed or missing fetches to the queue
-	if incomplete {
-		for _, header := range request.Headers[from+accepted : from+results] {
-			taskQueue.Push(header, -int64(header.Number.Uint64()))
-		}
-	} else {
-		for _, header := range request.Headers[from+accepted:] {
-			taskQueue.Push(header, -int64(header.Number.Uint64()))
-		}
+	for _, header := range request.Headers[accepted:] {
+		taskQueue.Push(header, -int64(header.Number.Uint64()))
 	}
 	// Wake up Results
 	if accepted > 0 {

eth/downloader/queue_test.go

@@ -32,45 +32,32 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/trie"
 )
 
-type blockConfig struct {
-	txPeriod int
-	txCount  int
-}
-
-var emptyBlock = blockConfig{txPeriod: 0, txCount: 0}
-var defaultBlock = blockConfig{txPeriod: 2, txCount: 1}
-
 // makeChain creates a chain of n blocks starting at and including parent.
 // The returned hash chain is ordered head->parent.
 // If empty is false, every second block (i%2==0) contains one transaction.
-// If config.txCount > 0, every config.txPeriod-th block contains config.txCount transactions.
 // No uncles are added.
-func makeChain(n int, seed byte, parent *types.Block, config blockConfig) ([]*types.Block, []types.Receipts) {
+func makeChain(n int, seed byte, parent *types.Block, empty bool) ([]*types.Block, []types.Receipts) {
 	blocks, receipts := core.GenerateChain(params.TestChainConfig, parent, ethash.NewFaker(), testDB, n, func(i int, block *core.BlockGen) {
 		block.SetCoinbase(common.Address{seed})
-		// Add transactions according to config
-		if config.txCount > 0 && i%config.txPeriod == 0 {
-			for range config.txCount {
-				signer := types.MakeSigner(params.TestChainConfig, block.Number(), block.Timestamp())
-				tx, err := types.SignTx(types.NewTransaction(block.TxNonce(testAddress), common.Address{seed}, big.NewInt(1000), params.TxGas, block.BaseFee(), nil), signer, testKey)
-				if err != nil {
-					panic(err)
-				}
-				block.AddTx(tx)
+		// Add one tx to every second block
+		if !empty && i%2 == 0 {
+			signer := types.MakeSigner(params.TestChainConfig, block.Number(), block.Timestamp())
+			tx, err := types.SignTx(types.NewTransaction(block.TxNonce(testAddress), common.Address{seed}, big.NewInt(1000), params.TxGas, block.BaseFee(), nil), signer, testKey)
+			if err != nil {
+				panic(err)
 			}
+			block.AddTx(tx)
 		}
 	})
 	return blocks, receipts
 }
 
 type chainData struct {
-	blocks   []*types.Block
-	receipts []types.Receipts
-	offset   int
+	blocks []*types.Block
+	offset int
 }
 
 var chain *chainData
@@ -79,11 +66,11 @@ var emptyChain *chainData
 func init() {
 	// Create a chain of blocks to import
 	targetBlocks := 128
-	blocks, receipts := makeChain(targetBlocks, 0, testGenesis, defaultBlock)
-	chain = &chainData{blocks, receipts, 0}
+	blocks, _ := makeChain(targetBlocks, 0, testGenesis, false)
+	chain = &chainData{blocks, 0}
 
-	blocks, receipts = makeChain(targetBlocks, 0, testGenesis, emptyBlock)
-	emptyChain = &chainData{blocks, receipts, 0}
+	blocks, _ = makeChain(targetBlocks, 0, testGenesis, true)
+	emptyChain = &chainData{blocks, 0}
 }
 
 func (chain *chainData) headers() []*types.Header {
@@ -274,149 +261,13 @@ func TestEmptyBlocks(t *testing.T) {
 	}
 }
 
-// TestPartialReceiptDelivery checks two points:
-// 1. Receipts that fail validation should be re-requested from other peers.
-// 2. Partial delivery should not expire.
-func TestPartialReceiptDelivery(t *testing.T) {
-	blocks, receipts := makeChain(64, 0, testGenesis, blockConfig{txPeriod: 1, txCount: 5})
-	chain := chainData{blocks: blocks, receipts: receipts, offset: 0}
-
-	numBlock := len(chain.blocks)
-
-	q := newQueue(10, 10)
-	if !q.Idle() {
-		t.Errorf("new queue should be idle")
-	}
-	q.Prepare(1, SnapSync)
-	if res := q.Results(false); len(res) != 0 {
-		t.Fatal("new queue should have 0 results")
-	}
-
-	// Schedule a batch of headers
-	headers := chain.headers()
-	hashes := make([]common.Hash, len(headers))
-	for i, header := range headers {
-		hashes[i] = header.Hash()
-	}
-	q.Schedule(headers, hashes, 1)
-
-	peer := dummyPeer("peer-1")
-	req, _, _ := q.ReserveReceipts(peer, numBlock)
-
-	t.Logf("request: length %d", len(req.Headers))
-
-	// 1. Deliver a partial receipt: this must not clear the remaining receipts from the pending list
-	firstCutoff := len(req.Headers) / 3
-	receiptRLP, rcHashes := getPartialReceiptsDelivery(0, firstCutoff, receipts)
-	accepted, err := q.DeliverReceipts(peer.id, receiptRLP, rcHashes, true, 0)
-	if err != nil || accepted != firstCutoff {
-		t.Fatalf("delivery failed: err %v, accepted %d\n", err, accepted)
-	}
-
-	if pending := q.PendingReceipts(); pending != numBlock-len(req.Headers) {
-		t.Fatalf("wrong pending receipt length, got %d, exp %d", pending, numBlock-len(req.Headers))
-	}
-	for i := range firstCutoff {
-		headerNumber := req.Headers[i].Number.Uint64()
-		res, _, _, _, err := q.resultCache.getFetchResult(headerNumber)
-		if err != nil {
-			t.Fatalf("fetch result get failed: err %v", err)
-		}
-		if res == nil {
-			t.Fatalf("fetch result is nil: header number %d", headerNumber)
-		}
-		if !res.Done(receiptType) {
-			t.Fatalf("wrong result, block %d receipt not done", headerNumber)
-		}
-	}
-	if flight := q.InFlightReceipts(); !flight {
-		t.Fatalf("there should be in flight receipts")
-	}
-
-	// 2. Deliver a partial receipt containing an invalid entry: the invalid receipt should be removed from the pending list
-	secondCutoff := firstCutoff + len(req.Headers)/3
-	receiptRLP, rcHashes = getPartialReceiptsDelivery(firstCutoff, secondCutoff, receipts)
-	// one invalid receipt
-	rcHashes[len(rcHashes)-1] = common.Hash{}
-	accepted, err = q.DeliverReceipts(peer.id, receiptRLP, rcHashes, true, firstCutoff)
-	if accepted != len(rcHashes)-1 {
-		t.Fatalf("wrong accepted, got %d, exp %d", accepted, len(rcHashes)-1)
-	}
-	if err == nil {
-		t.Fatalf("delivery should fail")
-	}
-
-	// The invalid receipt should be returned to the pending pool
-	if pending := q.PendingReceipts(); pending != numBlock-len(req.Headers)+1 {
-		t.Fatalf("wrong pending receipt length, got %d, exp %d", pending, numBlock-len(req.Headers))
-	}
-	for i := range len(rcHashes) - 1 {
-		headerNumber := req.Headers[firstCutoff+i].Number.Uint64()
-		res, _, _, _, err := q.resultCache.getFetchResult(headerNumber)
-		if err != nil {
-			t.Fatalf("fetch result get failed: err %v", err)
-		}
-		if res == nil {
-			t.Fatalf("fetch result is nil: header number %d", headerNumber)
-		}
-		if !res.Done(receiptType) {
-			t.Fatalf("wrong result, block %d receipt not done", headerNumber)
-		}
-	}
-
-	// 3. Deliver the remaining receipts to complete the request
-	thirdCutoff := len(req.Headers)
-	receiptRLP, rcHashes = getPartialReceiptsDelivery(secondCutoff, thirdCutoff, receipts)
-	accepted, err = q.DeliverReceipts(peer.id, receiptRLP, rcHashes, false, secondCutoff)
-	if accepted != len(rcHashes) {
-		t.Fatalf("wrong accepted, got %d, exp %d", accepted, len(rcHashes)-1)
-	}
-	if err != nil || accepted != thirdCutoff-secondCutoff {
-		t.Fatalf("delivery failed: err %v, accepted %d\n", err, accepted)
-	}
-
-	for i := range len(rcHashes) {
-		headerNumber := req.Headers[secondCutoff+i].Number.Uint64()
-		res, _, _, _, err := q.resultCache.getFetchResult(headerNumber)
-		if err != nil {
-			t.Fatalf("fetch result get failed: err %v", err)
-		}
-		if res == nil {
-			t.Fatalf("fetch result is nil: header number %d", headerNumber)
-		}
-		if !res.Done(receiptType) {
-			t.Fatalf("wrong result, block %d receipt not done", headerNumber)
-		}
-	}
-	if q.InFlightReceipts() {
-		t.Fatal("there shouldn't be any remaning in-flight receipts")
-	}
-}
-
-func getPartialReceiptsDelivery(from int, to int, receipts []types.Receipts) ([]rlp.RawValue, []common.Hash) {
-	if from < 0 {
-		from = 0
-	}
-	if to > len(receipts) {
-		to = len(receipts)
-	}
-
-	hasher := trie.NewStackTrie(nil)
-	rcHashes := make([]common.Hash, to-from)
-	for i, rc := range receipts[from:to] {
-		rcHashes[i] = types.DeriveSha(rc, hasher)
-	}
-
-	return types.EncodeBlockReceiptLists(receipts[from:to]), rcHashes
-}
-
 // XTestDelivery does some more extensive testing of events that happen,
 // blocks that become known and peers that make reservations and deliveries.
 // disabled since it's not really a unit-test, but can be executed to test
 // some more advanced scenarios
 func XTestDelivery(t *testing.T) {
 	// the outside network, holding blocks
-	blo, rec := makeChain(128, 0, testGenesis, defaultBlock)
+	blo, rec := makeChain(128, 0, testGenesis, false)
 	world := newNetwork()
 	world.receipts = rec
 	world.chain = blo
@@ -517,7 +368,7 @@ 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, false, 0)
+				_, err := q.DeliverReceipts(peer.id, types.EncodeBlockReceiptLists(rcs), hashes)
 				if err != nil {
 					fmt.Printf("delivered %d receipts %v\n", len(rcs), err)
 				}
@@ -593,7 +444,7 @@ func (n *network) progress(numBlocks int) {
 	n.lock.Lock()
 	defer n.lock.Unlock()
 	//fmt.Printf("progressing...\n")
-	newBlocks, newR := makeChain(numBlocks, 0, n.chain[len(n.chain)-1], emptyBlock)
+	newBlocks, newR := makeChain(numBlocks, 0, n.chain[len(n.chain)-1], false)
 	n.chain = append(n.chain, newBlocks...)
 	n.receipts = append(n.receipts, newR...)
 	n.cond.Broadcast()