eth/downloader: abort sync if master peer is unregistered

eth/downloader/downloader.go

@@ -314,6 +314,14 @@ func (d *Downloader) UnregisterPeer(id string) error {
 	}
 	d.queue.Revoke(id)
 
+	// If this peer was the master peer, abort sync immediately
+	d.cancelLock.RLock()
+	master := id == d.cancelPeer
+	d.cancelLock.RUnlock()
+
+	if master {
+		d.cancel()
+	}
 	return nil
 }
 
@@ -1275,21 +1283,14 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 							// Timeouts can occur if e.g. compaction hits at the wrong time, and can be ignored
 							peer.log.Warn("Downloader wants to drop peer, but peerdrop-function is not set", "peer", pid)
 						} else {
+							// In dropPeer function, a callback will be called which aborts
+							// the sync immediately. Here return the timeout error explicitly.
 							d.dropPeer(pid)
-
-							// If this peer was the master peer, abort sync immediately
-							d.cancelLock.RLock()
-							master := pid == d.cancelPeer
-							d.cancelLock.RUnlock()
-
-							if master {
-								d.cancel()
 							return errTimeout
 						}
 					}
 				}
 			}
-			}
 			// If there's nothing more to fetch, wait or terminate
 			if pending() == 0 {
 				if !inFlight() && finished {

eth/downloader/downloader_test.go

@@ -362,11 +362,20 @@ func (dl *downloadTester) dropPeer(id string) {
 	dl.downloader.UnregisterPeer(id)
 }
 
+// setDelay adds a response delay to test peer.
+func (dl *downloadTester) setDelay(id string, delay time.Duration) {
+	dl.lock.Lock()
+	defer dl.lock.Unlock()
+
+	dl.peers[id].delayResponse = delay
+}
+
 type downloadTesterPeer struct {
 	dl            *downloadTester
 	id            string
 	lock          sync.RWMutex
 	chain         *testChain
+	delayResponse time.Duration
 	missingStates map[common.Hash]bool // State entries that fast sync should not return
 }
 
@@ -384,7 +393,7 @@ func (dlp *downloadTesterPeer) RequestHeadersByHash(origin common.Hash, amount i
 	if reverse {
 		panic("reverse header requests not supported")
 	}
-
+	time.Sleep(dlp.delayResponse)
 	result := dlp.chain.headersByHash(origin, amount, skip)
 	go dlp.dl.downloader.DeliverHeaders(dlp.id, result)
 	return nil
@@ -397,7 +406,7 @@ func (dlp *downloadTesterPeer) RequestHeadersByNumber(origin uint64, amount int,
 	if reverse {
 		panic("reverse header requests not supported")
 	}
-
+	time.Sleep(dlp.delayResponse)
 	result := dlp.chain.headersByNumber(origin, amount, skip)
 	go dlp.dl.downloader.DeliverHeaders(dlp.id, result)
 	return nil
@@ -407,6 +416,7 @@ func (dlp *downloadTesterPeer) RequestHeadersByNumber(origin uint64, amount int,
 // peer in the download tester. The returned function can be used to retrieve
 // batches of block bodies from the particularly requested peer.
 func (dlp *downloadTesterPeer) RequestBodies(hashes []common.Hash) error {
+	time.Sleep(dlp.delayResponse)
 	txs, uncles := dlp.chain.bodies(hashes)
 	go dlp.dl.downloader.DeliverBodies(dlp.id, txs, uncles)
 	return nil
@@ -416,6 +426,7 @@ func (dlp *downloadTesterPeer) RequestBodies(hashes []common.Hash) error {
 // 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) error {
+	time.Sleep(dlp.delayResponse)
 	receipts := dlp.chain.receipts(hashes)
 	go dlp.dl.downloader.DeliverReceipts(dlp.id, receipts)
 	return nil
@@ -428,6 +439,7 @@ func (dlp *downloadTesterPeer) RequestNodeData(hashes []common.Hash) error {
 	dlp.dl.lock.RLock()
 	defer dlp.dl.lock.RUnlock()
 
+	time.Sleep(dlp.delayResponse)
 	results := make([][]byte, 0, len(hashes))
 	for _, hash := range hashes {
 		if data, err := dlp.dl.peerDb.Get(hash.Bytes()); err == nil {
@@ -1656,3 +1668,44 @@ func testCheckpointEnforcement(t *testing.T, protocol int, mode SyncMode) {
 		assertOwnChain(t, tester, chain.len())
 	}
 }
+
+func TestCancelMasterPeer62(t *testing.T)      { testCancelMasterPeer(t, 62, FullSync) }
+func TestCancelMasterPeer63Full(t *testing.T)  { testCancelMasterPeer(t, 63, FullSync) }
+func TestCancelMasterPeer63Fast(t *testing.T)  { testCancelMasterPeer(t, 63, FastSync) }
+func TestCancelMasterPeer64Full(t *testing.T)  { testCancelMasterPeer(t, 64, FullSync) }
+func TestCancelMasterPeer64Fast(t *testing.T)  { testCancelMasterPeer(t, 64, FastSync) }
+func TestCancelMasterPeer64Light(t *testing.T) { testCancelMasterPeer(t, 64, LightSync) }
+
+func testCancelMasterPeer(t *testing.T, protocol int, mode SyncMode) {
+	t.Parallel()
+
+	// Create a new tester
+	tester := newTester()
+	defer tester.terminate()
+
+	// Attempt to sync with the two peers
+	chain1 := testChainBase.shorten(MaxHeaderFetch + 64)
+	tester.newPeer("peer1", protocol, chain1) // Mark peer1 as the master peer
+
+	// Add some response delay so that we have enough time to unregister master
+	// peer before sync finished.
+	tester.setDelay("peer1", time.Duration(time.Millisecond*300))
+	chain2 := testChainBase.shorten(MaxHeaderFetch)
+	tester.newPeer("peer2", protocol, chain2) // Mark peer2 as the auxiliary peer
+
+	var errCh = make(chan error, 1)
+	go func() {
+		errCh <- tester.sync("peer1", nil, mode)
+	}()
+	time.Sleep(time.Millisecond * 300)        // Ensure we have started the syncing
+	tester.downloader.UnregisterPeer("peer1") // Unregister the master peer, which should abort sync
+
+	select {
+	case err := <-errCh:
+		if err != errCanceled {
+			t.Fatalf("error mismatch, want %v, got %v", errCanceled, err)
+		}
+	case <-time.NewTimer(time.Second).C:
+		t.Fatalf("timeout")
+	}
+}

eth/downloader/statesync.go

@@ -316,19 +316,12 @@ func (s *stateSync) loop() (err error) {
 					// Timeouts can occur if e.g. compaction hits at the wrong time, and can be ignored
 					req.peer.log.Warn("Downloader wants to drop peer, but peerdrop-function is not set", "peer", req.peer.id)
 				} else {
+					// In dropPeer function, a callback will be called which aborts
+					// the sync immediately. Here return the timeout error explicitly.
 					s.d.dropPeer(req.peer.id)
-
-					// If this peer was the master peer, abort sync immediately
-					s.d.cancelLock.RLock()
-					master := req.peer.id == s.d.cancelPeer
-					s.d.cancelLock.RUnlock()
-
-					if master {
-						s.d.cancel()
 					return errTimeout
 				}
 			}
-			}
 			// Process all the received blobs and check for stale delivery
 			delivered, err := s.process(req)
 			if err != nil {