provider: swarm exploration (#1120)

* provider: adding provide and reprovide queue

* provider: network operations

* add some tests

* schedule prefix len computations

* provider schedule

* provider: handleProvide

* addressed review

* use go-test/random

* satisfy linter

* log errors during initial prefix len measurement

* address review

* satisfy linter

* address review

* provider: explore swarm

* address review

* decrease minimal region size from replicationFactor+1 to replicationFactor

* simplify unscheduleSubsumedPrefixesNoClock

* address review

* refactor and test groupAndScheduleKeysByPrefix

* moved maxPrefixSize const to top

Author: guillaumemichel

provider/internal/keyspace/trie.go

@@ -305,9 +305,9 @@ type Region struct {
 	Keys   *trie.Trie[bit256.Key, mh.Multihash]
 }
 
-// RegionsFromPeers returns the keyspace regions of size `regionSize` from the
-// given `peers` sorted according to `order` along with the Common Prefix
-// shared by all peers.
+// RegionsFromPeers returns the keyspace regions of size at least `regionSize`
+// from the given `peers` sorted according to `order` along with the Common
+// Prefix shared by all peers.
 func RegionsFromPeers(peers []peer.ID, regionSize int, order bit256.Key) ([]Region, bitstr.Key) {
 	if len(peers) == 0 {
 		return []Region{}, ""
@@ -326,14 +326,14 @@ func RegionsFromPeers(peers []peer.ID, regionSize int, order bit256.Key) ([]Regi
 }
 
 // extractMinimalRegions returns the list of all non-overlapping subtries of
-// `t` having strictly more than `size` elements, sorted according to `order`.
-// Every element is included in exactly one region.
+// `t` having at least `size` elements, sorted according to `order`. Every
+// element is included in exactly one region.
 func extractMinimalRegions(t *trie.Trie[bit256.Key, peer.ID], path bitstr.Key, size int, order bit256.Key) []Region {
 	if t.IsEmptyLeaf() {
 		return nil
 	}
 	branch0, branch1 := t.Branch(0), t.Branch(1)
-	if branch0 != nil && branch1 != nil && branch0.Size() > size && branch1.Size() > size {
+	if branch0 != nil && branch1 != nil && branch0.Size() >= size && branch1.Size() >= size {
 		b := int(order.Bit(len(path)))
 		return append(extractMinimalRegions(t.Branch(b), path+bitstr.Key(byte('0'+b)), size, order),
 			extractMinimalRegions(t.Branch(1-b), path+bitstr.Key(byte('1'-b)), size, order)...)

provider/internal/keyspace/trie_test.go

@@ -580,15 +580,15 @@ func TestRegionsFromPeers(t *testing.T) {
 
 	// Two peers
 	p1 := genRandPeerID(t)
-	regions, commonPrefix = RegionsFromPeers([]peer.ID{p0, p1}, 1, bit256.ZeroKey())
+	regions, commonPrefix = RegionsFromPeers([]peer.ID{p0, p1}, 2, bit256.ZeroKey())
 	require.Len(t, regions, 1)
 	cpl := key.CommonPrefixLength(bstrPid0, PeerIDToBit256(p1))
 	common := bstrPid0[:cpl]
 	require.Equal(t, common, commonPrefix)
 
 	// Three peers
 	p2 := genRandPeerID(t)
-	regions, commonPrefix = RegionsFromPeers([]peer.ID{p0, p1, p2}, 1, bit256.ZeroKey())
+	regions, commonPrefix = RegionsFromPeers([]peer.ID{p0, p1, p2}, 2, bit256.ZeroKey())
 	require.Len(t, regions, 1)
 	cpl = key.CommonPrefixLength(common, PeerIDToBit256(p2))
 	common = common[:cpl]

provider/provider.go

@@ -2,6 +2,7 @@ package provider
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"sync"
@@ -98,6 +99,8 @@ type SweepingProvider struct {
 
 	keyStore *datastore.KeyStore
 
+	replicationFactor int
+
 	provideQueue   *queue.ProvideQueue
 	provideRunning sync.Mutex
 
@@ -129,7 +132,7 @@ type SweepingProvider struct {
 // FIXME: remove me
 func (s *SweepingProvider) SatisfyLinter() {
 	s.vanillaProvide([]byte{})
-	s.closestPeersToKey("")
+	s.exploreSwarm("")
 	s.measureInitialPrefixLen()
 }
 
@@ -396,6 +399,126 @@ func (s *SweepingProvider) vanillaProvide(k mh.Multihash) (bitstr.Key, error) {
 	return coveredPrefix, s.sendProviderRecords(keysAllocations, addrInfo)
 }
 
+// exploreSwarm finds all peers whose kademlia identifier matches `prefix` in
+// the DHT swarm, and organizes them in keyspace regions.
+//
+// A region is identified by a keyspace prefix, and contains all the peers
+// matching this prefix. A region always has at least s.replicationFactor
+// peers. Regions are non-overlapping.
+//
+// If there less than s.replicationFactor peers match `prefix`, explore
+// shorter prefixes until at least s.replicationFactor peers are included in
+// the region.
+//
+// The returned `coveredPrefix` represents the keyspace prefix covered by all
+// returned regions combined. It is different to the supplied `prefix` if there
+// aren't enough peers matching `prefix`.
+func (s *SweepingProvider) exploreSwarm(prefix bitstr.Key) (regions []keyspace.Region, coveredPrefix bitstr.Key, err error) {
+	peers, err := s.closestPeersToPrefix(prefix)
+	if err != nil {
+		return nil, "", fmt.Errorf("exploreSwarm '%s': %w", prefix, err)
+	}
+	if len(peers) == 0 {
+		return nil, "", fmt.Errorf("no peers found when exploring prefix %s", prefix)
+	}
+	regions, coveredPrefix = keyspace.RegionsFromPeers(peers, s.replicationFactor, s.order)
+	return regions, coveredPrefix, nil
+}
+
+// maxPrefixSearches is the maximum number of GetClosestPeers operations that
+// are allowed to explore a prefix, preventing an infinite loop, since the exit
+// condition depends on the network topology.
+//
+// A lower bound estimate on the number of fresh peers returned by GCP is
+// replicationFactor/2. Hence, 64 GCP are expected to return at least
+// 32*replicatonFactor peers, which should be more than enough, even if the
+// supplied prefix is too short.
+const maxPrefixSearches = 64
+
+// closestPeersToPrefix returns at least s.replicationFactor peers
+// corresponding to the branch of the network peers trie matching the provided
+// prefix. In the case there aren't enough peers matching the provided prefix,
+// it will find and return the closest peers to the prefix, even if they don't
+// exactly match it.
+func (s *SweepingProvider) closestPeersToPrefix(prefix bitstr.Key) ([]peer.ID, error) {
+	allClosestPeers := make(map[peer.ID]struct{})
+
+	nextPrefix := prefix
+	startTime := time.Now()
+	coveredPrefixesStack := []bitstr.Key{}
+
+	i := 0
+	// Go down the trie to fully cover prefix.
+exploration:
+	for {
+		if i == maxPrefixSearches {
+			return nil, errors.New("closestPeersToPrefix needed more than maxPrefixSearches iterations")
+		}
+		if !s.connectivity.IsOnline() {
+			return nil, errors.New("provider: node is offline")
+		}
+		i++
+		fullKey := keyspace.FirstFullKeyWithPrefix(nextPrefix, s.order)
+		closestPeers, err := s.closestPeersToKey(fullKey)
+		if err != nil {
+			// We only get an err if something really bad happened, e.g no peers in
+			// routing table, invalid key, etc.
+			return nil, err
+		}
+		if len(closestPeers) == 0 {
+			return nil, errors.New("dht lookup did not return any peers")
+		}
+		coveredPrefix, coveredPeers := keyspace.ShortestCoveredPrefix(fullKey, closestPeers)
+		for _, p := range coveredPeers {
+			allClosestPeers[p] = struct{}{}
+		}
+
+		coveredPrefixLen := len(coveredPrefix)
+		if i == 1 {
+			if coveredPrefixLen <= len(prefix) && coveredPrefix == prefix[:coveredPrefixLen] && len(allClosestPeers) >= s.replicationFactor {
+				// Exit early if the prefix is fully covered at the first request and
+				// we have enough (at least replicationFactor) peers.
+				break exploration
+			}
+		} else {
+			latestPrefix := coveredPrefixesStack[len(coveredPrefixesStack)-1]
+			for coveredPrefixLen <= len(latestPrefix) && coveredPrefix[:coveredPrefixLen-1] == latestPrefix[:coveredPrefixLen-1] {
+				// Pop latest prefix from stack, because current prefix is
+				// complementary.
+				// e.g latestPrefix=0010, currentPrefix=0011. latestPrefix is
+				// replaced by 001, unless 000 was also in the stack, etc.
+				coveredPrefixesStack = coveredPrefixesStack[:len(coveredPrefixesStack)-1]
+				coveredPrefix = coveredPrefix[:len(coveredPrefix)-1]
+				coveredPrefixLen = len(coveredPrefix)
+
+				if len(coveredPrefixesStack) == 0 {
+					if coveredPrefixLen <= len(prefix) && len(allClosestPeers) >= s.replicationFactor {
+						break exploration
+					}
+					// Not enough peers -> add coveredPrefix to stack and continue.
+					break
+				}
+				if coveredPrefixLen == 0 {
+					logger.Error("coveredPrefixLen==0, coveredPrefixStack ", coveredPrefixesStack)
+					break exploration
+				}
+				latestPrefix = coveredPrefixesStack[len(coveredPrefixesStack)-1]
+			}
+		}
+		// Push coveredPrefix to stack
+		coveredPrefixesStack = append(coveredPrefixesStack, coveredPrefix)
+		// Flip last bit of last covered prefix
+		nextPrefix = keyspace.FlipLastBit(coveredPrefixesStack[len(coveredPrefixesStack)-1])
+	}
+
+	peers := make([]peer.ID, 0, len(allClosestPeers))
+	for p := range allClosestPeers {
+		peers = append(peers, p)
+	}
+	logger.Debugf("Region %s exploration required %d requests to discover %d peers in %s", prefix, i, len(allClosestPeers), time.Since(startTime))
+	return peers, nil
+}
+
 // closestPeersToKey returns a valid peer ID sharing a long common prefix with
 // the provided key. Note that the returned peer IDs aren't random, they are
 // taken from a static list of preimages.