eth/fetcher: remove test sleeps (15s -> 2.8s)

2 files changed, 167 insertions, 59 deletions

diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go
index ceca79df0..7b5804ab2 100644
--- a/eth/fetcher/fetcher.go
+++ b/eth/fetcher/fetcher.go
@@ -92,6 +92,10 @@ type Fetcher struct {
 	chainHeight    chainHeightFn      // Retrieves the current chain's height
 	insertChain    chainInsertFn      // Injects a batch of blocks into the chain
 	dropPeer       peerDropFn         // Drops a peer for misbehaving
+
+	// Testing hooks
+	fetchingHook func([]common.Hash) // Method to call upon starting a block fetch
+	importedHook func(*types.Block)  // Method to call upon successful block import
 }
 
 // New creates a block fetcher to retrieve blocks based on hash announcements.
@@ -277,7 +281,13 @@ func (f *Fetcher) loop() {
 
 					glog.V(logger.Detail).Infof("Peer %s: fetching %s", peer, list)
 				}
-				go f.fetching[hashes[0]].fetch(hashes)
+				hashes := hashes // closure!
+				go func() {
+					if f.fetchingHook != nil {
+						f.fetchingHook(hashes)
+					}
+					f.fetching[hashes[0]].fetch(hashes)
+				}()
 			}
 			// Schedule the next fetch if blocks are still pending
 			f.reschedule(fetch)
@@ -402,6 +412,11 @@ func (f *Fetcher) insert(peer string, block *types.Block) {
 		}
 		// If import succeeded, broadcast the block
 		go f.broadcastBlock(block, false)
+
+		// Invoke the testing hook if needed
+		if f.importedHook != nil {
+			f.importedHook(block)
+		}
 	}()
 }
 
diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go
index b9f0f36a5..06ca2ef86 100644
--- a/eth/fetcher/fetcher_test.go
+++ b/eth/fetcher/fetcher_test.go
@@ -163,7 +163,7 @@ func (f *fetcherTester) makeFetcher(blocks map[common.Hash]*types.Block) blockRe
 // them, successfully importing into the local chain.
 func TestSequentialAnnouncements(t *testing.T) {
 	// Create a chain of blocks to import
-	targetBlocks := 24
+	targetBlocks := 4 * hashLimit
 	hashes := createHashes(targetBlocks, knownHash)
 	blocks := createBlocksFromHashes(hashes)
 
@@ -171,12 +171,22 @@ func TestSequentialAnnouncements(t *testing.T) {
 	fetcher := tester.makeFetcher(blocks)
 
 	// Iteratively announce blocks until all are imported
-	for i := len(hashes) - 1; i >= 0; i-- {
+	imported := make(chan *types.Block)
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+
+	for i := len(hashes) - 2; i >= 0; i-- {
 		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
-		time.Sleep(50 * time.Millisecond)
+
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", len(hashes)-i)
+		}
 	}
-	if imported := len(tester.blocks); imported != targetBlocks+1 {
-		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 }
 
@@ -184,7 +194,7 @@ func TestSequentialAnnouncements(t *testing.T) {
 // peer), they will only get downloaded at most once.
 func TestConcurrentAnnouncements(t *testing.T) {
 	// Create a chain of blocks to import
-	targetBlocks := 24
+	targetBlocks := 4 * hashLimit
 	hashes := createHashes(targetBlocks, knownHash)
 	blocks := createBlocksFromHashes(hashes)
 
@@ -198,15 +208,24 @@ func TestConcurrentAnnouncements(t *testing.T) {
 		return fetcher(hashes)
 	}
 	// Iteratively announce blocks until all are imported
-	for i := len(hashes) - 1; i >= 0; i-- {
+	imported := make(chan *types.Block)
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+
+	for i := len(hashes) - 2; i >= 0; i-- {
 		tester.fetcher.Notify("first", hashes[i], time.Now().Add(-arriveTimeout), wrapper)
 		tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper)
 		tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper)
 
-		time.Sleep(50 * time.Millisecond)
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", len(hashes)-i)
+		}
 	}
-	if imported := len(tester.blocks); imported != targetBlocks+1 {
-		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 	// Make sure no blocks were retrieved twice
 	if int(counter) != targetBlocks {
@@ -218,7 +237,7 @@ func TestConcurrentAnnouncements(t *testing.T) {
 // results in a valid import.
 func TestOverlappingAnnouncements(t *testing.T) {
 	// Create a chain of blocks to import
-	targetBlocks := 24
+	targetBlocks := 4 * hashLimit
 	hashes := createHashes(targetBlocks, knownHash)
 	blocks := createBlocksFromHashes(hashes)
 
@@ -226,15 +245,31 @@ func TestOverlappingAnnouncements(t *testing.T) {
 	fetcher := tester.makeFetcher(blocks)
 
 	// Iteratively announce blocks, but overlap them continuously
-	delay, overlap := 50*time.Millisecond, time.Duration(5)
-	for i := len(hashes) - 1; i >= 0; i-- {
-		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout+overlap*delay), fetcher)
-		time.Sleep(delay)
-	}
-	time.Sleep(overlap * delay)
+	fetching := make(chan []common.Hash)
+	imported := make(chan *types.Block, len(hashes)-1)
+	tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes }
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
 
-	if imported := len(tester.blocks); imported != targetBlocks+1 {
-		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
+	for i := len(hashes) - 2; i >= 0; i-- {
+		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
+		select {
+		case <-fetching:
+		case <-time.After(time.Second):
+			t.Fatalf("hash %d: announce timeout", len(hashes)-i)
+		}
+	}
+	// Wait for all the imports to complete and check count
+	for i := 0; i < len(hashes)-1; i++ {
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", i)
+		}
+	}
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 }
 
@@ -280,27 +315,37 @@ func TestPendingDeduplication(t *testing.T) {
 // imported when all the gaps are filled in.
 func TestRandomArrivalImport(t *testing.T) {
 	// Create a chain of blocks to import, and choose one to delay
-	targetBlocks := 24
-	hashes := createHashes(targetBlocks, knownHash)
+	hashes := createHashes(maxQueueDist, knownHash)
 	blocks := createBlocksFromHashes(hashes)
-	skip := targetBlocks / 2
+	skip := maxQueueDist / 2
 
 	tester := newTester()
 	fetcher := tester.makeFetcher(blocks)
 
 	// Iteratively announce blocks, skipping one entry
+	imported := make(chan *types.Block, len(hashes)-1)
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
 			tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
-			time.Sleep(50 * time.Millisecond)
+			time.Sleep(time.Millisecond)
 		}
 	}
 	// Finally announce the skipped entry and check full import
 	tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher)
-	time.Sleep(50 * time.Millisecond)
 
-	if imported := len(tester.blocks); imported != targetBlocks+1 {
-		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
+	for i := 0; i < len(hashes)-1; i++ {
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", i)
+		}
+	}
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 }
 
@@ -308,27 +353,37 @@ func TestRandomArrivalImport(t *testing.T) {
 // are correctly schedule, filling and import queue gaps.
 func TestQueueGapFill(t *testing.T) {
 	// Create a chain of blocks to import, and choose one to not announce at all
-	targetBlocks := 24
-	hashes := createHashes(targetBlocks, knownHash)
+	hashes := createHashes(maxQueueDist, knownHash)
 	blocks := createBlocksFromHashes(hashes)
-	skip := targetBlocks / 2
+	skip := maxQueueDist / 2
 
 	tester := newTester()
 	fetcher := tester.makeFetcher(blocks)
 
 	// Iteratively announce blocks, skipping one entry
+	imported := make(chan *types.Block, len(hashes)-1)
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
 			tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher)
-			time.Sleep(50 * time.Millisecond)
+			time.Sleep(time.Millisecond)
 		}
 	}
 	// Fill the missing block directly as if propagated
 	tester.fetcher.Enqueue("valid", blocks[hashes[skip]])
-	time.Sleep(50 * time.Millisecond)
 
-	if imported := len(tester.blocks); imported != targetBlocks+1 {
-		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1)
+	for i := 0; i < len(hashes)-1; i++ {
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", i)
+		}
+	}
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 }
 
@@ -348,9 +403,15 @@ func TestImportDeduplication(t *testing.T) {
 		atomic.AddUint32(&counter, uint32(len(blocks)))
 		return tester.insertChain(blocks)
 	}
+	// Instrument the fetching and imported events
+	fetching := make(chan []common.Hash)
+	imported := make(chan *types.Block, len(hashes)-1)
+	tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes }
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+
 	// Announce the duplicating block, wait for retrieval, and also propagate directly
 	tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher)
-	time.Sleep(50 * time.Millisecond)
+	<-fetching
 
 	tester.fetcher.Enqueue("valid", blocks[hashes[0]])
 	tester.fetcher.Enqueue("valid", blocks[hashes[0]])
@@ -358,8 +419,13 @@ func TestImportDeduplication(t *testing.T) {
 
 	// Fill the missing block directly as if propagated, and check import uniqueness
 	tester.fetcher.Enqueue("valid", blocks[hashes[1]])
-	time.Sleep(50 * time.Millisecond)
-
+	for done := false; !done; {
+		select {
+		case <-imported:
+		case <-time.After(50 * time.Millisecond):
+			done = true
+		}
+	}
 	if imported := len(tester.blocks); imported != 3 {
 		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 3)
 	}
@@ -400,8 +466,12 @@ func TestDistantDiscarding(t *testing.T) {
 // block announcements to a node, but that even in the face of such an attack,
 // the fetcher remains operational.
 func TestHashMemoryExhaustionAttack(t *testing.T) {
+	// Create a tester with instrumented import hooks
 	tester := newTester()
 
+	imported := make(chan *types.Block)
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+
 	// Create a valid chain and an infinite junk chain
 	hashes := createHashes(hashLimit+2*maxQueueDist, knownHash)
 	blocks := createBlocksFromHashes(hashes)
@@ -413,29 +483,39 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
 	// Feed the tester a huge hashset from the attacker, and a limited from the valid peer
 	for i := 0; i < len(attack); i++ {
 		if i < maxQueueDist {
-			tester.fetcher.Notify("valid", hashes[len(hashes)-1-i], time.Now().Add(arriveTimeout/2), valid)
+			tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], time.Now(), valid)
 		}
-		tester.fetcher.Notify("attacker", attack[i], time.Now().Add(arriveTimeout/2), attacker)
+		tester.fetcher.Notify("attacker", attack[i], time.Now(), attacker)
 	}
 	if len(tester.fetcher.announced) != hashLimit+maxQueueDist {
 		t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist)
 	}
-	// Wait for synchronisation to complete and check success for the valid peer
-	time.Sleep(2 * arriveTimeout)
-	if imported := len(tester.blocks); imported != maxQueueDist {
-		t.Fatalf("partial synchronised block mismatch: have %v, want %v", imported, maxQueueDist)
+	// Wait for fetches to complete
+	for i := 0; i < maxQueueDist; i++ {
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", i)
+		}
+	}
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 	// Feed the remaining valid hashes to ensure DOS protection state remains clean
-	for i := len(hashes) - maxQueueDist; i >= 0; {
-		for j := 0; j < maxQueueDist && i >= 0; j++ {
-			tester.fetcher.Notify("valid", hashes[i], time.Now().Add(time.Millisecond), valid)
-			i--
+	for i := len(hashes) - maxQueueDist - 2; i >= 0; i-- {
+		tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), valid)
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", len(hashes)-i)
 		}
-		time.Sleep(500 * time.Millisecond)
 	}
-	time.Sleep(500 * time.Millisecond)
-	if imported := len(tester.blocks); imported != len(hashes) {
-		t.Fatalf("fully synchronised block mismatch: have %v, want %v", imported, len(hashes))
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 }
 
@@ -443,14 +523,18 @@ func TestHashMemoryExhaustionAttack(t *testing.T) {
 // announces and retrievals) don't pile up indefinitely, exhausting available
 // system memory.
 func TestBlockMemoryExhaustionAttack(t *testing.T) {
+	// Create a tester with instrumented import hooks
 	tester := newTester()
 
+	imported := make(chan *types.Block)
+	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+
 	// Create a valid chain and a batch of dangling (but in range) blocks
-	hashes := createHashes(blockLimit, knownHash)
+	hashes := createHashes(blockLimit+2*maxQueueDist, knownHash)
 	blocks := createBlocksFromHashes(hashes)
 
 	attack := make(map[common.Hash]*types.Block)
-	for i := 0; i < 16; i++ {
+	for len(attack) < blockLimit+2*maxQueueDist {
 		hashes := createHashes(maxQueueDist-1, unknownHash)
 		blocks := createBlocksFromHashes(hashes)
 		for _, hash := range hashes[:maxQueueDist-2] {
@@ -475,18 +559,27 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) {
 	}
 	// Insert the missing piece (and sanity check the import)
 	tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2]])
-	time.Sleep(500 * time.Millisecond)
-	if imported := len(tester.blocks); imported != maxQueueDist+1 {
-		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, maxQueueDist+1)
+	for i := 0; i < maxQueueDist; i++ {
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", i)
+		}
+	}
+	select {
+	case <-imported:
+		t.Fatalf("extra block imported")
+	case <-time.After(50 * time.Millisecond):
 	}
 	// Insert the remaining blocks in chunks to ensure clean DOS protection
 	for i := maxQueueDist; i < len(hashes)-1; i++ {
 		tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2-i]])
-		if i%maxQueueDist == 0 {
-			time.Sleep(500 * time.Millisecond)
+		select {
+		case <-imported:
+		case <-time.After(time.Second):
+			t.Fatalf("block %d: import timeout", len(hashes)-i)
 		}
 	}
-	time.Sleep(500 * time.Millisecond)
 	if imported := len(tester.blocks); imported != len(hashes) {
 		t.Fatalf("synchronised block mismatch: have %v, want %v", imported, len(hashes))
 	}