Test query interactions with routing table

v2/coord/coordinator.go

@@ -150,7 +150,7 @@
 
 func NewCoordinator(self kadt.PeerID, rtr Router, 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
 	}
@@ -158,8 +158,8 @@
 	// 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
@@ -241,13 +241,13 @@
 	return nil
 }
 
 func (c *Coordinator) ID() kadt.PeerID {
 	return c.self
 }
 
 func (c *Coordinator) Addresses() []ma.Multiaddr {
 	// TODO: return configured listen addresses
 	info, err := c.rtr.GetNodeInfo(context.TODO(), peer.ID(c.self))
 	if err != nil {
 		return nil
 	}
@@ -308,6 +308,8 @@
 // GetNode retrieves the node associated with the given node id from the DHT's local routing table.
 // If the node isn't found in the table, it returns ErrNodeNotFound.
 func (c *Coordinator) GetNode(ctx context.Context, id peer.ID) (Node, error) {
+	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.GetNode")
+	defer span.End()
 	if _, exists := c.rt.GetNode(kadt.PeerID(id).Key()); !exists {
 		return nil, ErrNodeNotFound
 	}
@@ -321,6 +323,8 @@
 
 // GetClosestNodes requests the n closest nodes to the key from the node's local routing table.
 func (c *Coordinator) GetClosestNodes(ctx context.Context, k kadt.Key, n int) ([]Node, error) {
+	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.GetClosestNodes")
+	defer span.End()
 	closest := c.rt.NearestNodes(k, n)
 	nodes := make([]Node, 0, len(closest))
 	for _, id := range closest {
@@ -335,7 +339,7 @@
 
 // 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) (Value, error) {
 	panic("not implemented")
 }
 
@@ -462,3 +466,32 @@
 
 	return nil
 }
+
+// NotifyConnectivity notifies the coordinator that a peer has passed a connectivity check
+// which means it is connected and supports finding closer nodes
+func (c *Coordinator) NotifyConnectivity(ctx context.Context, id peer.ID) error {
+	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.NotifyConnectivity")
+	defer span.End()
+
+	ai := peer.AddrInfo{
+		ID: id,
+	}
+	c.routingBehaviour.Notify(ctx, &EventNotifyConnectivity{
+		NodeInfo: ai,
+	})
+
+	return nil
+}
+
+// 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 peer.ID) error {
+	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.NotifyNonConnectivity")
+	defer span.End()
+
+	c.routingBehaviour.Notify(ctx, &EventNotifyNonConnectivity{
+		NodeID: id,
+	})
+
+	return nil
+}

v2/coord/event.go

@@ -86,26 +86,32 @@
 func (*EventStopQuery) behaviourEvent() {}
 func (*EventStopQuery) queryCommand()   {}
 
+// EventAddAddrInfo notifies the routing behaviour of a potential new peer or of additional addresses for
+// an existing peer.
 type EventAddAddrInfo struct {
 	NodeInfo peer.AddrInfo
 }
 
 func (*EventAddAddrInfo) behaviourEvent() {}
 func (*EventAddAddrInfo) routingCommand() {}
 
+// EventGetCloserNodesSuccess notifies a behaviour that a GetCloserNodes request, initiated by an
+// [EventOutboundGetCloserNodes] event has produced a successful response.
 type EventGetCloserNodesSuccess struct {
 	QueryID     query.QueryID
-	To          peer.AddrInfo
+	To          peer.AddrInfo // To is the peer address that the GetCloserNodes request was sent to.
 	Target      kadt.Key
 	CloserNodes []peer.AddrInfo
 }
 
 func (*EventGetCloserNodesSuccess) behaviourEvent()      {}
 func (*EventGetCloserNodesSuccess) nodeHandlerResponse() {}
 
+// EventGetCloserNodesFailure notifies a behaviour that a GetCloserNodes request, initiated by an
+// [EventOutboundGetCloserNodes] event has failed to produce a valid response.
 type EventGetCloserNodesFailure struct {
 	QueryID query.QueryID
-	To      peer.AddrInfo
+	To      peer.AddrInfo // To is the peer address that the GetCloserNodes request was sent to.
 	Target  kadt.Key
 	Err     error
 }
@@ -141,6 +147,14 @@
 func (*EventRoutingUpdated) behaviourEvent()      {}
 func (*EventRoutingUpdated) routingNotification() {}
 
+// EventRoutingRemoved is emitted by the coordinator when new node has been removed from the routing table.
+type EventRoutingRemoved struct {
+	NodeID peer.ID
+}
+
+func (*EventRoutingRemoved) behaviourEvent()      {}
+func (*EventRoutingRemoved) routingNotification() {}
+
 // EventBootstrapFinished is emitted by the coordinator when a bootstrap has finished, either through
 // running to completion or by being canceled.
 type EventBootstrapFinished struct {
@@ -149,3 +163,23 @@
 
 func (*EventBootstrapFinished) behaviourEvent()      {}
 func (*EventBootstrapFinished) routingNotification() {}
+
+// EventNotifyConnectivity notifies a behaviour that a peer's connectivity and support for finding closer nodes
+// has been confirmed such as from a successful query response or an inbound query. This should not be used for
+// general connections to the host but only when it is confirmed that the peer responds to requests for closer
+// nodes.
+type EventNotifyConnectivity struct {
+	NodeInfo peer.AddrInfo
+}
+
+func (*EventNotifyConnectivity) behaviourEvent()      {}
+func (*EventNotifyConnectivity) routingNotification() {}
+
+// EventNotifyNonConnectivity notifies a behaviour that a peer does not have connectivity and/or does not support
+// finding closer nodes is known.
+type EventNotifyNonConnectivity struct {
+	NodeID peer.ID
+}
+
+func (*EventNotifyNonConnectivity) behaviourEvent() {}
+func (*EventNotifyNonConnectivity) routingCommand() {}

v2/coord/query.go

@@ -81,6 +81,11 @@ func (p *PooledQueryBehaviour) Notify(ctx context.Context, ev BehaviourEvent) {
 			CloserNodes: sliceOfAddrInfoToSliceOfKadPeerID(ev.CloserNodes),
 		}
 	case *EventGetCloserNodesFailure:
+		// queue an event that will notify the routing behaviour of a failed node
+		p.pending = append(p.pending, &EventNotifyNonConnectivity{
+			ev.To.ID,
+		})
+
 		cmd = &query.EventPoolFindCloserFailure[kadt.Key, kadt.PeerID]{
 			NodeID:  kadt.PeerID(ev.To.ID),
 			QueryID: ev.QueryID,

v2/coord/query/query.go

@@ -93,15 +93,15 @@
 
 func NewQuery[K kad.Key[K], N kad.NodeID[K]](self N, id QueryID, target K, iter NodeIter[K, N], knownClosestNodes []N, cfg *QueryConfig[K]) (*Query[K, N], error) {
 	if cfg == nil {
 		cfg = DefaultQueryConfig[K]()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	for _, node := range knownClosestNodes {
 		// exclude self from closest nodes
 		if key.Equal(node.Key(), self.Key()) {
 			continue
 		}
 		iter.Add(&NodeStatus[K, N]{
 			NodeID: node,
@@ -130,19 +130,23 @@
 
 	switch tev := ev.(type) {
 	case *EventQueryCancel:
+		span.SetAttributes(tele.AttrEvent("EventQueryCancel"))
 		q.markFinished()
 		return &StateQueryFinished{
 			QueryID: q.id,
 			Stats:   q.stats,
 		}
 	case *EventQueryFindCloserResponse[K, N]:
+		span.SetAttributes(tele.AttrEvent("EventQueryFindCloserResponse"))
 		q.onMessageResponse(ctx, tev.NodeID, tev.CloserNodes)
 	case *EventQueryFindCloserFailure[K, N]:
+		span.SetAttributes(tele.AttrEvent("EventQueryFindCloserFailure"))
+		span.RecordError(tev.Error)
 		q.onMessageFailure(ctx, tev.NodeID)
 	case nil:
 		// TEMPORARY: no event to process
 	default:
 		panic(fmt.Sprintf("unexpected event: %T", tev))
 	}
 
 	// count number of successes in the order of the iteration
@@ -170,6 +174,7 @@
 				q.inFlight--
 				q.stats.Failure++
 			} else if atCapacity() {
+				span.SetAttributes(tele.AttrOutEvent("StateQueryWaitingAtCapacity")) // this is the query's tracing span
 				returnState = &StateQueryWaitingAtCapacity{
 					QueryID: q.id,
 					Stats:   q.stats,
@@ -186,6 +191,7 @@
 			// If it has contacted at least NumResults nodes successfully then the iteration is done.
 			if !progressing && successes >= q.cfg.NumResults {
 				q.markFinished()
+				span.SetAttributes(tele.AttrOutEvent("StateQueryFinished")) // this is the query's tracing span
 				returnState = &StateQueryFinished{
 					QueryID: q.id,
 					Stats:   q.stats,
@@ -202,6 +208,7 @@
 				if q.stats.Start.IsZero() {
 					q.stats.Start = q.cfg.Clock.Now()
 				}
+				span.SetAttributes(tele.AttrOutEvent("StateQueryFindCloser")) // this is the query's tracing span
 				returnState = &StateQueryFindCloser[K, N]{
 					NodeID:  ni.NodeID,
 					QueryID: q.id,
@@ -211,17 +218,18 @@
 				return true
 
 			}
+			span.SetAttributes(tele.AttrOutEvent("StateQueryWaitingAtCapacity")) // this is the query's tracing span
 			returnState = &StateQueryWaitingAtCapacity{
 				QueryID: q.id,
 				Stats:   q.stats,
 			}
 			return true
 		case *StateNodeUnresponsive:
 			// ignore
 		case *StateNodeFailed:
 			// ignore
 		default:
 			panic(fmt.Sprintf("unexpected state: %T", ni.State))
 		}
 
 		return false
@@ -233,6 +241,7 @@
 
 	if q.inFlight > 0 {
 		// The iterator is still waiting for results and not at capacity
+		span.SetAttributes(tele.AttrOutEvent("StateQueryWaitingWithCapacity"))
 		return &StateQueryWaitingWithCapacity{
 			QueryID: q.id,
 			Stats:   q.stats,
@@ -242,6 +251,7 @@
 	// The iterator is finished because all available nodes have been contacted
 	// and the iterator is not waiting for any more results.
 	q.markFinished()
+	span.SetAttributes(tele.AttrOutEvent("StateQueryFinished"))
 	return &StateQueryFinished{
 		QueryID: q.id,
 		Stats:   q.stats,
@@ -266,20 +276,20 @@
 	case *StateNodeWaiting:
 		q.inFlight--
 		q.stats.Success++
 	case *StateNodeUnresponsive:
 		q.stats.Success++
 
 	case *StateNodeNotContacted:
 		// ignore duplicate or late response
 		return
 	case *StateNodeFailed:
 		// ignore duplicate or late response
 		return
 	case *StateNodeSucceeded:
 		// ignore duplicate or late response
 		return
 	default:
 		panic(fmt.Sprintf("unexpected state: %T", st))
 	}
 
 	// add closer nodes to list
@@ -300,27 +310,27 @@
 func (q *Query[K, N]) onMessageFailure(ctx context.Context, node N) {
 	ni, found := q.iter.Find(node.Key())
 	if !found {
 		// got a rogue message
 		return
 	}
 	switch st := ni.State.(type) {
 	case *StateNodeWaiting:
 		q.inFlight--
 		q.stats.Failure++
 	case *StateNodeUnresponsive:
 		// update node state to failed
 		break
 	case *StateNodeNotContacted:
 		// update node state to failed
 		break
 	case *StateNodeFailed:
 		// ignore duplicate or late response
 		return
 	case *StateNodeSucceeded:
 		// ignore duplicate or late response
 		return
 	default:
 		panic(fmt.Sprintf("unexpected state: %T", st))
 	}
 
 	ni.State = &StateNodeFailed{}
@@ -357,10 +367,10 @@
 }
 
 // queryState() ensures that only [Query] states can be assigned to a QueryState.
 func (*StateQueryFinished) queryState()            {}
 func (*StateQueryFindCloser[K, N]) queryState()    {}
 func (*StateQueryWaitingAtCapacity) queryState()   {}
 func (*StateQueryWaitingWithCapacity) queryState() {}
 
 type QueryEvent interface {
 	queryEvent()
@@ -382,6 +392,6 @@
 }
 
 // queryEvent() ensures that only events accepted by [Query] can be assigned to a [QueryEvent].
 func (*EventQueryCancel) queryEvent()                   {}
 func (*EventQueryFindCloserResponse[K, N]) queryEvent() {}
 func (*EventQueryFindCloserFailure[K, N]) queryEvent()  {}

v2/coord/routing.go

@@ -91,7 +91,7 @@
 		}
 
 	case *EventRoutingUpdated:
-		span.SetAttributes(attribute.String("event", "EventRoutingUpdated"))
+		span.SetAttributes(attribute.String("event", "EventRoutingUpdated"), attribute.String("nodeid", ev.NodeInfo.ID.String()))
 		cmd := &routing.EventProbeAdd[kadt.Key, kadt.PeerID]{
 			NodeID: addrInfoToKadPeerID(ev.NodeInfo),
 		}
@@ -130,10 +130,10 @@
 					NodeID: kadt.PeerID(ev.To.ID),
 				}
 			} else {
 				cmd = &routing.EventIncludeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 					NodeID: kadt.PeerID(ev.To.ID),
 					Error:  fmt.Errorf("response did not include any closer nodes"),
 				}
 			}
 			// attempt to advance the include
 			next, ok := r.advanceInclude(ctx, cmd)
@@ -142,17 +142,17 @@
 			}
 
 		case "probe":
 			var cmd routing.ProbeEvent
 			// require that the node responded with at least one closer node
 			if len(ev.CloserNodes) > 0 {
 				cmd = &routing.EventProbeConnectivityCheckSuccess[kadt.Key, kadt.PeerID]{
 					NodeID: kadt.PeerID(ev.To.ID),
 				}
 			} else {
 				cmd = &routing.EventProbeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 					NodeID: kadt.PeerID(ev.To.ID),
 					Error:  fmt.Errorf("response did not include any closer nodes"),
 				}
 			}
 			// attempt to advance the probe state machine
 			next, ok := r.advanceProbe(ctx, cmd)
@@ -188,9 +188,9 @@
 				r.pending = append(r.pending, next)
 			}
 		case "probe":
 			cmd := &routing.EventProbeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 				NodeID: kadt.PeerID(ev.To.ID),
 				Error:  ev.Err,
 			}
 			// attempt to advance the probe state machine
 			next, ok := r.advanceProbe(ctx, cmd)
@@ -201,6 +201,41 @@
 		default:
 			panic(fmt.Sprintf("unexpected query id: %s", ev.QueryID))
 		}
+	case *EventNotifyConnectivity:
+		span.SetAttributes(attribute.String("event", "EventNotifyConnectivity"), attribute.String("nodeid", ev.NodeInfo.ID.String()))
+		// ignore self
+		if ev.NodeInfo.ID == peer.ID(r.self) {
+			break
+		}
+		// tell the include state machine in case this is a new peer that could be added to the routing table
+		cmd := &routing.EventIncludeAddCandidate[kadt.Key, kadt.PeerID]{
+			NodeID: kadt.PeerID(ev.NodeInfo.ID),
+		}
+		next, ok := r.advanceInclude(ctx, cmd)
+		if ok {
+			r.pending = append(r.pending, next)
+		}
+
+		// tell the probe state machine in case there is are connectivity checks that could satisfied
+		cmdProbe := &routing.EventProbeNotifyConnectivity[kadt.Key, kadt.PeerID]{
+			NodeID: kadt.PeerID(ev.NodeInfo.ID),
+		}
+		nextProbe, ok := r.advanceProbe(ctx, cmdProbe)
+		if ok {
+			r.pending = append(r.pending, nextProbe)
+		}
+	case *EventNotifyNonConnectivity:
+		span.SetAttributes(attribute.String("event", "EventNotifyConnectivity"), attribute.String("nodeid", ev.NodeID.String()))
+
+		// tell the probe state machine to remove the node from the routing table and probe list
+		cmdProbe := &routing.EventProbeRemove[kadt.Key, kadt.PeerID]{
+			NodeID: kadt.PeerID(ev.NodeID),
+		}
+		nextProbe, ok := r.advanceProbe(ctx, cmdProbe)
+		if ok {
+			r.pending = append(r.pending, nextProbe)
+		}
+
 	default:
 		panic(fmt.Sprintf("unexpected dht event: %T", ev))
 	}
@@ -351,7 +386,13 @@
 			Notify:  r,
 		}, true
 	case *routing.StateProbeNodeFailure[kadt.Key, kadt.PeerID]:
-		// a node has failed a connectivity check been removed from the routing table and the probe list
+		// a node has failed a connectivity check and been removed from the routing table and the probe list
+
+		// emit an EventRoutingRemoved event to notify clients that the node has been removed
+		r.pending = append(r.pending, &EventRoutingRemoved{
+			NodeID: peer.ID(st.NodeID),
+		})
+
 		// add the node to the inclusion list for a second chance
 		r.notify(ctx, &EventAddAddrInfo{
 			NodeInfo: kadPeerIDToAddrInfo(st.NodeID),

v2/coord/routing/bootstrap.go

@@ -8,6 +8,7 @@
 	"github.com/benbjohnson/clock"
 	"github.com/plprobelab/go-kademlia/kad"
 	"github.com/plprobelab/go-kademlia/kaderr"
+	"go.opentelemetry.io/otel/attribute"
 
 	"github.com/libp2p/go-libp2p-kad-dht/v2/coord/query"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/tele"
@@ -78,10 +79,10 @@
 
 func NewBootstrap[K kad.Key[K], N kad.NodeID[K]](self N, cfg *BootstrapConfig[K]) (*Bootstrap[K, N], error) {
 	if cfg == nil {
 		cfg = DefaultBootstrapConfig[K]()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	return &Bootstrap[K, N]{
 		self: self,
@@ -96,6 +97,7 @@
 
 	switch tev := ev.(type) {
 	case *EventBootstrapStart[K, N]:
+		span.SetAttributes(tele.AttrEvent("EventBootstrapStart"))
 
 		// TODO: ignore start event if query is already in progress
 		iter := query.NewClosestNodesIter[K, N](b.self.Key())
@@ -109,27 +111,31 @@
 
 		qry, err := query.NewQuery[K, N](b.self, queryID, b.self.Key(), iter, tev.KnownClosestNodes, qryCfg)
 		if err != nil {
 			// TODO: don't panic
 			panic(err)
 		}
 		b.qry = qry
 		return b.advanceQuery(ctx, nil)
 
 	case *EventBootstrapFindCloserResponse[K, N]:
+		span.SetAttributes(tele.AttrEvent("EventBootstrapFindCloserResponse"))
 		return b.advanceQuery(ctx, &query.EventQueryFindCloserResponse[K, N]{
 			NodeID:      tev.NodeID,
 			CloserNodes: tev.CloserNodes,
 		})
 	case *EventBootstrapFindCloserFailure[K, N]:
+		span.SetAttributes(tele.AttrEvent("EventBootstrapFindCloserFailure"))
+		span.RecordError(tev.Error)
 		return b.advanceQuery(ctx, &query.EventQueryFindCloserFailure[K, N]{
 			NodeID: tev.NodeID,
 			Error:  tev.Error,
 		})
 
 	case *EventBootstrapPoll:
+		span.SetAttributes(tele.AttrEvent("EventBootstrapPoll"))
 	// ignore, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected event: %T", tev))
 	}
 
 	if b.qry != nil {
@@ -140,41 +146,49 @@
 }
 
 func (b *Bootstrap[K, N]) advanceQuery(ctx context.Context, qev query.QueryEvent) BootstrapState {
+	ctx, span := tele.StartSpan(ctx, "Bootstrap.advanceQuery")
+	defer span.End()
 	state := b.qry.Advance(ctx, qev)
 	switch st := state.(type) {
 	case *query.StateQueryFindCloser[K, N]:
+		span.SetAttributes(attribute.String("out_state", "StateQueryFindCloser"))
 		return &StateBootstrapFindCloser[K, N]{
 			QueryID: st.QueryID,
 			Stats:   st.Stats,
 			NodeID:  st.NodeID,
 			Target:  st.Target,
 		}
 	case *query.StateQueryFinished:
+		span.SetAttributes(attribute.String("out_state", "StateBootstrapFinished"))
 		return &StateBootstrapFinished{
 			Stats: st.Stats,
 		}
 	case *query.StateQueryWaitingAtCapacity:
 		elapsed := b.cfg.Clock.Since(st.Stats.Start)
 		if elapsed > b.cfg.Timeout {
+			span.SetAttributes(attribute.String("out_state", "StateBootstrapTimeout"))
 			return &StateBootstrapTimeout{
 				Stats: st.Stats,
 			}
 		}
+		span.SetAttributes(attribute.String("out_state", "StateBootstrapWaiting"))
 		return &StateBootstrapWaiting{
 			Stats: st.Stats,
 		}
 	case *query.StateQueryWaitingWithCapacity:
 		elapsed := b.cfg.Clock.Since(st.Stats.Start)
 		if elapsed > b.cfg.Timeout {
+			span.SetAttributes(attribute.String("out_state", "StateBootstrapTimeout"))
 			return &StateBootstrapTimeout{
 				Stats: st.Stats,
 			}
 		}
+		span.SetAttributes(attribute.String("out_state", "StateBootstrapWaiting"))
 		return &StateBootstrapWaiting{
 			Stats: st.Stats,
 		}
 	default:
 		panic(fmt.Sprintf("unexpected state: %T", st))
 	}
 }
 
@@ -210,11 +224,11 @@
 }
 
 // bootstrapState() ensures that only Bootstrap states can be assigned to a BootstrapState.
 func (*StateBootstrapFindCloser[K, N]) bootstrapState() {}
 func (*StateBootstrapIdle) bootstrapState()             {}
 func (*StateBootstrapFinished) bootstrapState()         {}
 func (*StateBootstrapTimeout) bootstrapState()          {}
 func (*StateBootstrapWaiting) bootstrapState()          {}
 
 // BootstrapEvent is an event intended to advance the state of a bootstrap.
 type BootstrapEvent interface {
@@ -242,7 +256,7 @@
 }
 
 // bootstrapEvent() ensures that only events accepted by a [Bootstrap] can be assigned to the [BootstrapEvent] interface.
 func (*EventBootstrapPoll) bootstrapEvent()                     {}
 func (*EventBootstrapStart[K, N]) bootstrapEvent()              {}
 func (*EventBootstrapFindCloserResponse[K, N]) bootstrapEvent() {}
 func (*EventBootstrapFindCloserFailure[K, N]) bootstrapEvent()  {}