Add explore state machine to expand population of routing table

v2/go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/libp2p/go-msgio v0.3.0
 	github.com/multiformats/go-base32 v0.1.0
 	github.com/multiformats/go-multiaddr v0.11.0
-	github.com/multiformats/go-multihash v0.2.3 // indirect
+	github.com/multiformats/go-multihash v0.2.3
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/plprobelab/go-kademlia v0.0.0-20230913171354-443ec1f56080
 	github.com/prometheus/client_golang v1.16.0 // indirect

v2/internal/coord/behaviour_test.go

@@ -4,13 +4,6 @@ import (
 	"context"
 )
 
-type NullSM[E any, S any] struct{}
-
-func (NullSM[E, S]) Advance(context.Context, E) S {
-	var v S
-	return v
-}
-
 type RecordingSM[E any, S any] struct {
 	State    S
 	Received E

v2/internal/coord/coordinator.go

@@ -22,6 +22,7 @@
 
 	"github.com/libp2p/go-libp2p-kad-dht/v2/internal/coord/brdcst"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/internal/coord/coordt"
+	"github.com/libp2p/go-libp2p-kad-dht/v2/internal/coord/cplutil"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/internal/coord/query"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/internal/coord/routing"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/kadt"
@@ -169,16 +170,16 @@
 
 func NewCoordinator(self kadt.PeerID, rtr coordt.Router[kadt.Key, kadt.PeerID, *pb.Message], rt routing.RoutingTableCpl[kadt.Key, kadt.PeerID], cfg *CoordinatorConfig) (*Coordinator, error) {
 	if cfg == nil {
 		cfg = DefaultCoordinatorConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	// initialize a new telemetry struct
 	tele, err := NewTelemetry(cfg.MeterProvider, cfg.TracerProvider)
 	if err != nil {
 		return nil, fmt.Errorf("init telemetry: %w", err)
 	}
 
 	qpCfg := query.DefaultPoolConfig()
 	qpCfg.Clock = cfg.Clock
@@ -189,8 +190,8 @@
 
 	qp, err := query.NewPool[kadt.Key, kadt.PeerID, *pb.Message](self, qpCfg)
 	if err != nil {
 		return nil, fmt.Errorf("query pool: %w", err)
 	}
 	queryBehaviour := NewPooledQueryBehaviour(qp, cfg.Logger, tele.Tracer)
 
 	bootstrapCfg := routing.DefaultBootstrapConfig[kadt.Key]()
@@ -201,8 +202,8 @@
 
 	bootstrap, err := routing.NewBootstrap(self, bootstrapCfg)
 	if err != nil {
 		return nil, fmt.Errorf("bootstrap: %w", err)
 	}
 
 	includeCfg := routing.DefaultIncludeConfig()
 	includeCfg.Clock = cfg.Clock
@@ -215,8 +216,8 @@
 
 	include, err := routing.NewInclude[kadt.Key, kadt.PeerID](rt, includeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("include: %w", err)
 	}
 
 	probeCfg := routing.DefaultProbeConfig()
 	probeCfg.Clock = cfg.Clock
@@ -226,17 +227,32 @@
 	// probeCfg.Concurrency = cfg.ProbeConcurrency
 	probe, err := routing.NewProbe[kadt.Key](rt, probeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("probe: %w", err)
 	}
 
-	routingBehaviour := NewRoutingBehaviour(self, bootstrap, include, probe, cfg.Logger, tele.Tracer)
+	exploreCfg := routing.DefaultExploreConfig()
+	exploreCfg.Clock = cfg.Clock
+	exploreCfg.Timeout = cfg.QueryTimeout
+
+	schedule, err := routing.NewDynamicExploreSchedule(14, cfg.Clock.Now(), time.Hour, 1, 0)
+	if err != nil {
+		return nil, fmt.Errorf("explore schedule: %w", err)
+	}
+
+	// TODO: expose more config
+	explore, err := routing.NewExplore[kadt.Key](self, rt, cplutil.GenRandPeerID, schedule, exploreCfg)
+	if err != nil {
+		return nil, fmt.Errorf("explore: %w", err)
+	}
+
+	routingBehaviour := NewRoutingBehaviour(self, bootstrap, include, probe, explore, cfg.Logger, tele.Tracer)
 
 	networkBehaviour := NewNetworkBehaviour(rtr, cfg.Logger, tele.Tracer)
 
 	b, err := brdcst.NewPool[kadt.Key, kadt.PeerID, *pb.Message](self, nil)
 	if err != nil {
 		return nil, fmt.Errorf("broadcast: %w", err)
 	}
 
 	brdcstBehaviour := NewPooledBroadcastBehaviour(b, cfg.Logger, tele.Tracer)
 
@@ -271,8 +287,8 @@
 	return nil
 }
 
 func (c *Coordinator) ID() kadt.PeerID {
 	return c.self
 }
 
 func (c *Coordinator) eventLoop(ctx context.Context) {
@@ -289,14 +305,14 @@
 		case <-ctx.Done():
 			// coordinator is closing
 			return
 		case <-c.networkBehaviour.Ready():
 			ev, ok = c.networkBehaviour.Perform(ctx)
 		case <-c.routingBehaviour.Ready():
 			ev, ok = c.routingBehaviour.Perform(ctx)
 		case <-c.queryBehaviour.Ready():
 			ev, ok = c.queryBehaviour.Perform(ctx)
 		case <-c.brdcstBehaviour.Ready():
 			ev, ok = c.brdcstBehaviour.Perform(ctx)
 		}
 
 		if ok {
@@ -312,10 +328,10 @@
 	switch ev := ev.(type) {
 	case NetworkCommand:
 		c.networkBehaviour.Notify(ctx, ev)
 	case QueryCommand:
 		c.queryBehaviour.Notify(ctx, ev)
 	case BrdcstCommand:
 		c.brdcstBehaviour.Notify(ctx, ev)
 	case RoutingCommand:
 		c.routingBehaviour.Notify(ctx, ev)
 	case RoutingNotification:
@@ -323,8 +339,8 @@
 		rn := c.routingNotifier
 		c.routingNotifierMu.RUnlock()
 		rn.Notify(ctx, ev)
 	default:
 		panic(fmt.Sprintf("unexpected event: %T", ev))
 	}
 }
 
@@ -345,8 +361,8 @@
 
 	nh, err := c.networkBehaviour.getNodeHandler(ctx, id)
 	if err != nil {
 		return nil, err
 	}
 	return nh, nil
 }
 
@@ -359,8 +375,8 @@
 	for _, id := range closest {
 		nh, err := c.networkBehaviour.getNodeHandler(ctx, id)
 		if err != nil {
 			return nil, err
 		}
 		nodes = append(nodes, nh)
 	}
 	return nodes, nil
@@ -368,14 +384,14 @@
 
 // GetValue requests that the node return any value associated with the supplied key.
 // If the node does not have a value for the key it returns ErrValueNotFound.
 func (c *Coordinator) GetValue(ctx context.Context, k kadt.Key) (coordt.Value, error) {
 	panic("not implemented")
 }
 
 // PutValue requests that the node stores a value to be associated with the supplied key.
 // If the node cannot or chooses not to store the value for the key it returns ErrValueNotAccepted.
 func (c *Coordinator) PutValue(ctx context.Context, r coordt.Value, q int) error {
 	panic("not implemented")
 }
 
 // QueryClosest starts a query that attempts to find the closest nodes to the target key.
@@ -398,8 +414,8 @@
 
 	seeds, err := c.GetClosestNodes(ctx, target, 20)
 	if err != nil {
 		return nil, coordt.QueryStats{}, err
 	}
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
@@ -442,13 +458,13 @@
 	defer cancel()
 
 	if numResults < 1 {
 		numResults = 20
 	}
 
 	seeds, err := c.GetClosestNodes(ctx, msg.Target(), numResults)
 	if err != nil {
 		return coordt.QueryStats{}, err
 	}
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
@@ -474,51 +490,51 @@
 	return stats, err
 }
 
 func (c *Coordinator) BroadcastRecord(ctx context.Context, msg *pb.Message) error {
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.BroadcastRecord")
 	defer span.End()
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	seeds, err := c.GetClosestNodes(ctx, msg.Target(), 20)
 	if err != nil {
 		return err
 	}
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
 		seedIDs = append(seedIDs, s.ID())
 	}
 
 	waiter := NewWaiter[BehaviourEvent]()
 	queryID := c.newOperationID()
 
 	cmd := &EventStartBroadcast{
 		QueryID: queryID,
 		Target:  msg.Target(),
 		Message: msg,
 		Seed:    seedIDs,
 		Notify:  waiter,
 		Config:  brdcst.DefaultConfigFollowUp(),
 	}
 
 	// queue the start of the query
 	c.brdcstBehaviour.Notify(ctx, cmd)
 
 	contacted, errs, err := c.waitForBroadcast(ctx, waiter)
 	fmt.Println(contacted)
 	fmt.Println(errs)
 
 	return err
 }
 
 func (c *Coordinator) waitForQuery(ctx context.Context, queryID coordt.QueryID, waiter *Waiter[BehaviourEvent], fn coordt.QueryFunc) ([]kadt.PeerID, coordt.QueryStats, error) {
 	var lastStats coordt.QueryStats
 	for {
 		select {
 		case <-ctx.Done():
 			return nil, lastStats, ctx.Err()
 		case wev := <-waiter.Chan():
 			ctx, ev := wev.Ctx, wev.Event
 			switch ev := ev.(type) {
@@ -531,8 +547,8 @@
 				}
 				nh, err := c.networkBehaviour.getNodeHandler(ctx, ev.NodeID)
 				if err != nil {
 					// ignore unknown node
 					break
 				}
 
 				err = fn(ctx, nh.ID(), ev.Response, lastStats)
@@ -542,18 +558,18 @@
 					return nil, lastStats, nil
 				}
 				if err != nil {
 					// user defined error that terminates the query
 					c.queryBehaviour.Notify(ctx, &EventStopQuery{QueryID: queryID})
 					return nil, lastStats, err
 				}
 
 			case *EventQueryFinished:
 				// query is done
 				lastStats.Exhausted = true
 				return ev.ClosestNodes, lastStats, nil
 
 			default:
 				panic(fmt.Sprintf("unexpected event: %T", ev))
 			}
 		}
 	}
@@ -563,19 +579,19 @@
 	Node kadt.PeerID
 	Err  error
 }, error,
 ) {
 	for {
 		select {
 		case <-ctx.Done():
 			return nil, nil, ctx.Err()
 		case wev := <-waiter.Chan():
 			switch ev := wev.Event.(type) {
 			case *EventQueryProgressed:
 			case *EventBroadcastFinished:
 				return ev.Contacted, ev.Errors, nil
 
 			default:
 				panic(fmt.Sprintf("unexpected event: %T", ev))
 			}
 		}
 	}
@@ -589,8 +605,8 @@
 	defer span.End()
 	for _, id := range ids {
 		if id.Equal(c.self) {
 			// skip self
 			continue
 		}
 
 		c.routingBehaviour.Notify(ctx, &EventAddNode{
@@ -629,15 +645,15 @@
 
 // NotifyNonConnectivity notifies the coordinator that a peer has failed a connectivity check
 // which means it is not connected and/or it doesn't support finding closer nodes
 func (c *Coordinator) NotifyNonConnectivity(ctx context.Context, id kadt.PeerID) error {
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.NotifyNonConnectivity")
 	defer span.End()
 
 	c.routingBehaviour.Notify(ctx, &EventNotifyNonConnectivity{
 		NodeID: id,
 	})
 
 	return nil
 }
 
 func (c *Coordinator) newOperationID() coordt.QueryID {
@@ -685,8 +701,8 @@
 
 		// wait to be signaled that there is a new event
 		select {
 		case <-ctx.Done():
 			return nil, fmt.Errorf("test deadline exceeded while waiting for event %T", expected)
 		case <-w.signal:
 		}
 	}
@@ -711,8 +727,8 @@
 
 		// wait to be signaled that there is a new event
 		select {
 		case <-ctx.Done():
 			return nil, fmt.Errorf("test deadline exceeded while waiting for routing updated event")
 		case <-w.signal:
 		}
 	}
@@ -737,8 +753,8 @@
 
 		// wait to be signaled that there is a new event
 		select {
 		case <-ctx.Done():
 			return nil, fmt.Errorf("test deadline exceeded while waiting for routing removed event")
 		case <-w.signal:
 		}
 	}

v2/internal/coord/coordinator_test.go

@@ -99,11 +99,11 @@ func TestExhaustiveQuery(t *testing.T) {
 	// A (ids[0]) is looking for D (ids[3])
 	// A will first ask B, B will reply with C's address (and A's address)
 	// A will then ask C, C will reply with D's address (and B's address)
-	self := kadt.PeerID(nodes[0].NodeID)
+	self := nodes[0].NodeID
 	c, err := NewCoordinator(self, nodes[0].Router, nodes[0].RoutingTable, ccfg)
 	require.NoError(t, err)
 
-	target := kadt.PeerID(nodes[3].NodeID).Key()
+	target := nodes[3].NodeID.Key()
 
 	visited := make(map[string]int)
 
@@ -137,7 +137,7 @@ func TestRoutingUpdatedEventEmittedForCloserNodes(t *testing.T) {
 	// A (ids[0]) is looking for D (ids[3])
 	// A will first ask B, B will reply with C's address (and A's address)
 	// A will then ask C, C will reply with D's address (and B's address)
-	self := kadt.PeerID(nodes[0].NodeID)
+	self := nodes[0].NodeID
 	c, err := NewCoordinator(self, nodes[0].Router, nodes[0].RoutingTable, ccfg)
 	if err != nil {
 		log.Fatalf("unexpected error creating coordinator: %v", err)
@@ -194,7 +194,7 @@ func TestBootstrap(t *testing.T) {
 
 	ccfg.Clock = clk
 
-	self := kadt.PeerID(nodes[0].NodeID)
+	self := nodes[0].NodeID
 	d, err := NewCoordinator(self, nodes[0].Router, nodes[0].RoutingTable, ccfg)
 	require.NoError(t, err)
 

v2/internal/coord/cplutil/cpl.go

@@ -0,0 +1,58 @@
+package cplutil
+
+import (
+	"crypto/rand"
+	"encoding/binary"
+	"fmt"
+
+	mh "github.com/multiformats/go-multihash"
+
+	"github.com/libp2p/go-libp2p-kad-dht/v2/kadt"
+)
+
+//go:generate go run ./gen.go
+
+// GenRandPeerID generates a random [kadt.PeerID] whose key has a common prefix length of exactly cpl with the supplied key.
+// Ported from go-libp2p-kbucket
+func GenRandPeerID(k kadt.Key, cpl int) (kadt.PeerID, error) {
+	if cpl > 15 {
+		return "", fmt.Errorf("cannot generate peer ID for Cpl greater than 15")
+	}
+
+	targetPrefix := prefix(k, cpl)
+
+	// Convert to a known peer ID.
+	key := keyPrefixMap[targetPrefix]
+	id := [32 + 2]byte{mh.SHA2_256, 32}
+	binary.BigEndian.PutUint32(id[2:], key)
+	return kadt.PeerID(string(id[:])), nil
+}
+
+type keybit interface {
+	Bit(i int) uint
+}
+
+// prefix generates random bits that have a common prefix length of exactly cpl with the supplied key.
+func prefix(k keybit, cpl int) uint16 {
+	var p uint16
+	// copy the first cpl+1 bits so we can flip the last one
+	for i := 0; i < cpl+1; i++ {
+		bit := uint16(k.Bit(i)) << (15 - i)
+		p |= bit
+	}
+
+	// flip the bit at cpl (cpl 5 means bits 0-4 must be the same)
+	mask := uint16(1) << (15 - cpl)
+	p ^= mask
+
+	if cpl < 15 {
+		// pad with random data
+		var buf [2]byte
+		_, _ = rand.Read(buf[:])
+		r := binary.BigEndian.Uint16(buf[:])
+
+		mask = (^uint16(0)) << (15 - cpl)
+		p = (p & mask) | (r & ^mask)
+	}
+	return p
+}