Merge pull request #1859 from fjl/fix-discover-refresh-race

p2p/discover: fix race involving the seed node iterator

Author: obscuren

p2p/discover/database.go

@@ -21,6 +21,7 @@ package discover
 
 import (
 	"bytes"
+	"crypto/rand"
 	"encoding/binary"
 	"os"
 	"sync"
@@ -46,11 +47,8 @@ var (
 
 // nodeDB stores all nodes we know about.
 type nodeDB struct {
-	lvl    *leveldb.DB       // Interface to the database itself
-	seeder iterator.Iterator // Iterator for fetching possible seed nodes
-
-	self NodeID // Own node id to prevent adding it into the database
-
+	lvl    *leveldb.DB   // Interface to the database itself
+	self   NodeID        // Own node id to prevent adding it into the database
 	runner sync.Once     // Ensures we can start at most one expirer
 	quit   chan struct{} // Channel to signal the expiring thread to stop
 }
@@ -302,52 +300,70 @@ func (db *nodeDB) updateFindFails(id NodeID, fails int) error {
 	return db.storeInt64(makeKey(id, nodeDBDiscoverFindFails), int64(fails))
 }
 
-// querySeeds retrieves a batch of nodes to be used as potential seed servers
-// during bootstrapping the node into the network.
-//
-// Ideal seeds are the most recently seen nodes (highest probability to be still
-// alive), but yet untried. However, since leveldb only supports dumb iteration
-// we will instead start pulling in potential seeds that haven't been yet pinged
-// since the start of the boot procedure.
-//
-// If the database runs out of potential seeds, we restart the startup counter
-// and start iterating over the peers again.
-func (db *nodeDB) querySeeds(n int) []*Node {
-	// Create a new seed iterator if none exists
-	if db.seeder == nil {
-		db.seeder = db.lvl.NewIterator(nil, nil)
+// querySeeds retrieves random nodes to be used as potential seed nodes
+// for bootstrapping.
+func (db *nodeDB) querySeeds(n int, maxAge time.Duration) []*Node {
+	var (
+		now   = time.Now()
+		nodes = make([]*Node, 0, n)
+		it    = db.lvl.NewIterator(nil, nil)
+		id    NodeID
+	)
+	defer it.Release()
+
+seek:
+	for seeks := 0; len(nodes) < n && seeks < n*5; seeks++ {
+		// Seek to a random entry. The first byte is incremented by a
+		// random amount each time in order to increase the likelihood
+		// of hitting all existing nodes in very small databases.
+		ctr := id[0]
+		rand.Read(id[:])
+		id[0] = ctr + id[0]%16
+		it.Seek(makeKey(id, nodeDBDiscoverRoot))
+
+		n := nextNode(it)
+		if n == nil {
+			id[0] = 0
+			continue seek // iterator exhausted
+		}
+		if n.ID == db.self {
+			continue seek
+		}
+		if now.Sub(db.lastPong(n.ID)) > maxAge {
+			continue seek
+		}
+		for i := range nodes {
+			if nodes[i].ID == n.ID {
+				continue seek // duplicate
+			}
+		}
+		nodes = append(nodes, n)
 	}
-	// Iterate over the nodes and find suitable seeds
-	nodes := make([]*Node, 0, n)
-	for len(nodes) < n && db.seeder.Next() {
-		// Iterate until a discovery node is found
-		id, field := splitKey(db.seeder.Key())
+	return nodes
+}
+
+// reads the next node record from the iterator, skipping over other
+// database entries.
+func nextNode(it iterator.Iterator) *Node {
+	for end := false; !end; end = !it.Next() {
+		id, field := splitKey(it.Key())
 		if field != nodeDBDiscoverRoot {
 			continue
 		}
-		// Dump it if its a self reference
-		if bytes.Compare(id[:], db.self[:]) == 0 {
-			db.deleteNode(id)
+		var n Node
+		if err := rlp.DecodeBytes(it.Value(), &n); err != nil {
+			if glog.V(logger.Warn) {
+				glog.Errorf("invalid node %x: %v", id, err)
+			}
 			continue
 		}
-		// Load it as a potential seed
-		if node := db.node(id); node != nil {
-			nodes = append(nodes, node)
-		}
-	}
-	// Release the iterator if we reached the end
-	if len(nodes) == 0 {
-		db.seeder.Release()
-		db.seeder = nil
+		return &n
 	}
-	return nodes
+	return nil
 }
 
 // close flushes and closes the database files.
 func (db *nodeDB) close() {
-	if db.seeder != nil {
-		db.seeder.Release()
-	}
 	close(db.quit)
 	db.lvl.Close()
 }

p2p/discover/database_test.go

@@ -162,9 +162,33 @@ var nodeDBSeedQueryNodes = []struct {
 	node *Node
 	pong time.Time
 }{
+	// This one should not be in the result set because its last
+	// pong time is too far in the past.
 	{
 		node: newNode(
-			MustHexID("0x01d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
+			MustHexID("0x84d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
+			net.IP{127, 0, 0, 3},
+			30303,
+			30303,
+		),
+		pong: time.Now().Add(-3 * time.Hour),
+	},
+	// This one shouldn't be in in the result set because its
+	// nodeID is the local node's ID.
+	{
+		node: newNode(
+			MustHexID("0x57d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
+			net.IP{127, 0, 0, 3},
+			30303,
+			30303,
+		),
+		pong: time.Now().Add(-4 * time.Second),
+	},
+
+	// These should be in the result set.
+	{
+		node: newNode(
+			MustHexID("0x22d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
 			net.IP{127, 0, 0, 1},
 			30303,
 			30303,
@@ -173,7 +197,7 @@ var nodeDBSeedQueryNodes = []struct {
 	},
 	{
 		node: newNode(
-			MustHexID("0x02d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
+			MustHexID("0x44d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
 			net.IP{127, 0, 0, 2},
 			30303,
 			30303,
@@ -182,7 +206,7 @@ var nodeDBSeedQueryNodes = []struct {
 	},
 	{
 		node: newNode(
-			MustHexID("0x03d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
+			MustHexID("0xe2d9d65c4552b5eb43d5ad55a2ee3f56c6cbc1c64a5c8d659f51fcd51bace24351232b8d7821617d2b29b54b81cdefb9b3e9c37d7fd5f63270bcc9e1a6f6a439"),
 			net.IP{127, 0, 0, 3},
 			30303,
 			30303,
@@ -192,28 +216,32 @@ var nodeDBSeedQueryNodes = []struct {
 }
 
 func TestNodeDBSeedQuery(t *testing.T) {
-	db, _ := newNodeDB("", Version, NodeID{})
+	db, _ := newNodeDB("", Version, nodeDBSeedQueryNodes[1].node.ID)
 	defer db.close()
 
 	// Insert a batch of nodes for querying
 	for i, seed := range nodeDBSeedQueryNodes {
 		if err := db.updateNode(seed.node); err != nil {
 			t.Fatalf("node %d: failed to insert: %v", i, err)
 		}
+		if err := db.updateLastPong(seed.node.ID, seed.pong); err != nil {
+			t.Fatalf("node %d: failed to insert lastPong: %v", i, err)
+		}
 	}
+
 	// Retrieve the entire batch and check for duplicates
-	seeds := db.querySeeds(2 * len(nodeDBSeedQueryNodes))
-	if len(seeds) != len(nodeDBSeedQueryNodes) {
-		t.Errorf("seed count mismatch: have %v, want %v", len(seeds), len(nodeDBSeedQueryNodes))
-	}
+	seeds := db.querySeeds(len(nodeDBSeedQueryNodes)*2, time.Hour)
 	have := make(map[NodeID]struct{})
 	for _, seed := range seeds {
 		have[seed.ID] = struct{}{}
 	}
 	want := make(map[NodeID]struct{})
-	for _, seed := range nodeDBSeedQueryNodes {
+	for _, seed := range nodeDBSeedQueryNodes[2:] {
 		want[seed.node.ID] = struct{}{}
 	}
+	if len(seeds) != len(want) {
+		t.Errorf("seed count mismatch: have %v, want %v", len(seeds), len(want))
+	}
 	for id, _ := range have {
 		if _, ok := want[id]; !ok {
 			t.Errorf("extra seed: %v", id)
@@ -224,63 +252,6 @@ func TestNodeDBSeedQuery(t *testing.T) {
 			t.Errorf("missing seed: %v", id)
 		}
 	}
-	// Make sure the next batch is empty (seed EOF)
-	seeds = db.querySeeds(2 * len(nodeDBSeedQueryNodes))
-	if len(seeds) != 0 {
-		t.Errorf("seed count mismatch: have %v, want %v", len(seeds), 0)
-	}
-}
-
-func TestNodeDBSeedQueryContinuation(t *testing.T) {
-	db, _ := newNodeDB("", Version, NodeID{})
-	defer db.close()
-
-	// Insert a batch of nodes for querying
-	for i, seed := range nodeDBSeedQueryNodes {
-		if err := db.updateNode(seed.node); err != nil {
-			t.Fatalf("node %d: failed to insert: %v", i, err)
-		}
-	}
-	// Iteratively retrieve the batch, checking for an empty batch on reset
-	for i := 0; i < len(nodeDBSeedQueryNodes); i++ {
-		if seeds := db.querySeeds(1); len(seeds) != 1 {
-			t.Errorf("1st iteration %d: seed count mismatch: have %v, want %v", i, len(seeds), 1)
-		}
-	}
-	if seeds := db.querySeeds(1); len(seeds) != 0 {
-		t.Errorf("reset: seed count mismatch: have %v, want %v", len(seeds), 0)
-	}
-	for i := 0; i < len(nodeDBSeedQueryNodes); i++ {
-		if seeds := db.querySeeds(1); len(seeds) != 1 {
-			t.Errorf("2nd iteration %d: seed count mismatch: have %v, want %v", i, len(seeds), 1)
-		}
-	}
-}
-
-func TestNodeDBSelfSeedQuery(t *testing.T) {
-	// Assign a node as self to verify evacuation
-	self := nodeDBSeedQueryNodes[0].node.ID
-	db, _ := newNodeDB("", Version, self)
-	defer db.close()
-
-	// Insert a batch of nodes for querying
-	for i, seed := range nodeDBSeedQueryNodes {
-		if err := db.updateNode(seed.node); err != nil {
-			t.Fatalf("node %d: failed to insert: %v", i, err)
-		}
-	}
-	// Retrieve the entire batch and check that self was evacuated
-	seeds := db.querySeeds(2 * len(nodeDBSeedQueryNodes))
-	if len(seeds) != len(nodeDBSeedQueryNodes)-1 {
-		t.Errorf("seed count mismatch: have %v, want %v", len(seeds), len(nodeDBSeedQueryNodes)-1)
-	}
-	have := make(map[NodeID]struct{})
-	for _, seed := range seeds {
-		have[seed.ID] = struct{}{}
-	}
-	if _, ok := have[self]; ok {
-		t.Errorf("self not evacuated")
-	}
 }
 
 func TestNodeDBPersistency(t *testing.T) {