WIP

v2/internal/coord/coordinator.go

@@ -266,9 +266,11 @@ func (c *Coordinator) eventLoop(ctx context.Context) {
 
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.eventLoop")
 	defer span.End()
+
 	for {
 		var ev BehaviourEvent
 		var ok bool
+
 		select {
 		case <-ctx.Done():
 			// coordinator is closing
@@ -383,7 +385,7 @@ func (c *Coordinator) QueryClosest(ctx context.Context, target kadt.Key, fn Quer
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
-		seedIDs = append(seedIDs, kadt.PeerID(s.ID()))
+		seedIDs = append(seedIDs, s.ID())
 	}
 
 	waiter := NewWaiter[BehaviourEvent]()
@@ -432,7 +434,7 @@ func (c *Coordinator) QueryMessage(ctx context.Context, msg *pb.Message, fn Quer
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
-		seedIDs = append(seedIDs, kadt.PeerID(s.ID()))
+		seedIDs = append(seedIDs, s.ID())
 	}
 
 	waiter := NewWaiter[BehaviourEvent]()

v2/internal/coord/network.go

@@ -51,21 +51,19 @@ func (b *NetworkBehaviour) Notify(ctx context.Context, ev BehaviourEvent) {
 	switch ev := ev.(type) {
 	case *EventOutboundGetCloserNodes:
 		b.nodeHandlersMu.Lock()
-		p := kadt.PeerID(ev.To)
-		nh, ok := b.nodeHandlers[p]
+		nh, ok := b.nodeHandlers[ev.To]
 		if !ok {
-			nh = NewNodeHandler(p, b.rtr, b.logger, b.tracer)
-			b.nodeHandlers[p] = nh
+			nh = NewNodeHandler(ev.To, b.rtr, b.logger, b.tracer)
+			b.nodeHandlers[ev.To] = nh
 		}
 		b.nodeHandlersMu.Unlock()
 		nh.Notify(ctx, ev)
 	case *EventOutboundSendMessage:
 		b.nodeHandlersMu.Lock()
-		p := kadt.PeerID(ev.To)
-		nh, ok := b.nodeHandlers[p]
+		nh, ok := b.nodeHandlers[ev.To]
 		if !ok {
-			nh = NewNodeHandler(p, b.rtr, b.logger, b.tracer)
-			b.nodeHandlers[p] = nh
+			nh = NewNodeHandler(ev.To, b.rtr, b.logger, b.tracer)
+			b.nodeHandlers[ev.To] = nh
 		}
 		b.nodeHandlersMu.Unlock()
 		nh.Notify(ctx, ev)

v2/internal/coord/query.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/libp2p/go-libp2p-kad-dht/v2/kadt"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/pb"
+	"github.com/libp2p/go-libp2p-kad-dht/v2/tele"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/exp/slog"
 
@@ -189,9 +190,12 @@ func (p *PooledQueryBehaviour) Perform(ctx context.Context) (BehaviourEvent, boo
 	}
 }
 
-func (p *PooledQueryBehaviour) advancePool(ctx context.Context, ev query.PoolEvent) (BehaviourEvent, bool) {
-	ctx, span := p.tracer.Start(ctx, "PooledQueryBehaviour.advancePool")
-	defer span.End()
+func (p *PooledQueryBehaviour) advancePool(ctx context.Context, ev query.PoolEvent) (out BehaviourEvent, term bool) {
+	ctx, span := p.tracer.Start(ctx, "PooledQueryBehaviour.advancePool", trace.WithAttributes(tele.AttrInEvent(ev)))
+	defer func() {
+		span.SetAttributes(tele.AttrOutEvent(out))
+		span.End()
+	}()
 
 	pstate := p.pool.Advance(ctx, ev)
 	switch st := pstate.(type) {

v2/routing.go

@@ -113,6 +113,11 @@ func (d *DHT) FindProvidersAsync(ctx context.Context, c cid.Cid, count int) <-ch
 	panic("implement me")
 }
 
+// PutValue satisfies the [routing.Routing] interface and will add the given
+// value to the k-closest nodes to keyStr. The parameter keyStr should have the
+// format `/$namespace/$binary_id`. Namespace examples are `pk` or `ipns`. To
+// identify the closest peers to keyStr, that complete string will be SHA256
+// hashed.
 func (d *DHT) PutValue(ctx context.Context, keyStr string, value []byte, opts ...routing.Option) error {
 	ctx, span := d.tele.Tracer.Start(ctx, "DHT.PutValue")
 	defer span.End()
@@ -133,18 +138,25 @@ func (d *DHT) PutValue(ctx context.Context, keyStr string, value []byte, opts ..
 		return nil
 	}
 
-	fn := func(ctx context.Context, node coord.Node, stats coord.QueryStats) error {
-		return nil
-	}
-
+	// construct Kademlia-key. Yes, we hash the complete key string which
+	// includes the namespace prefix.
 	h := sha256.Sum256([]byte(keyStr))
 	kadKey := key.NewKey256(h[:])
 
-	_, err := d.kad.Query(ctx, kadKey, fn)
+	// define the query function that will be called after each request to a
+	// remote peer.
+	fn := func(ctx context.Context, id kadt.PeerID, resp *pb.Message, stats coord.QueryStats) error {
+		return nil
+	}
+
+	// finally, find the closest peers to the target key.
+	closest, _, err := d.kad.QueryClosest(ctx, kadKey, fn, 20)
 	if err != nil {
 		return fmt.Errorf("query error: %w", err)
 	}
 
+	_ = closest
+
 	panic("implement me")
 }