refactor: v2 simplify tracing

v2/config.go

@@ -8,6 +8,7 @@ import (
 	ds "github.com/ipfs/go-datastore"
 	leveldb "github.com/ipfs/go-ds-leveldb"
 	logging "github.com/ipfs/go-log/v2"
+	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	ma "github.com/multiformats/go-multiaddr"
 	manet "github.com/multiformats/go-multiaddr/net"
@@ -27,9 +28,9 @@ import (
 const ServiceName = "libp2p.DHT"
 
 const (
-	// ProtocolIPFS is the protocol identifier for the main IPFS network. If the
-	// DHT is configured with this protocol, you must configure backends for
-	// IPNS, Public Key, and provider records (ipns, pk, and providers
+	// ProtocolIPFS is the protocol identifier for the main Amino DHT network.
+	// If the DHT is configured with this protocol, you must configure backends
+	// for IPNS, Public Key, and provider records (ipns, pk, and providers
 	// namespaces). Configuration validation will fail if backends are missing.
 	ProtocolIPFS protocol.ID = "/ipfs/kad/1.0.0"
 
@@ -117,6 +118,10 @@ type Config struct {
 	// BucketSize determines the number of closer peers to return
 	BucketSize int
 
+	// BootstrapPeers is the list of peers that should be used to bootstrap
+	// into the DHT network.
+	BootstrapPeers []peer.AddrInfo
+
 	// ProtocolID represents the DHT [protocol] we can query with and respond to.
 	//
 	// [protocol]: https://docs.libp2p.io/concepts/fundamentals/protocols/
@@ -167,10 +172,16 @@ type Config struct {
 	// used to filter out private addresses.
 	AddressFilter AddressFilter
 
-	// MeterProvider .
+	// MeterProvider provides access to named Meter instances. It's used to,
+	// e.g., expose prometheus metrics. Check out the [opentelemetry docs]:
+	//
+	// [opentelemetry docs]: https://opentelemetry.io/docs/specs/otel/metrics/api/#meterprovider
 	MeterProvider metric.MeterProvider
 
-	// TracerProvider .
+	// TracerProvider provides Tracers that are used by instrumentation code to
+	// trace computational workflows. Check out the [opentelemetry docs]:
+	//
+	// [opentelemetry docs]: https://opentelemetry.io/docs/concepts/signals/traces/#tracer-provider
 	TracerProvider trace.TracerProvider
 }
 
@@ -184,6 +195,7 @@ func DefaultConfig() *Config {
 		Mode:              ModeOptAutoClient,
 		Kademlia:          coord.DefaultCoordinatorConfig(),
 		BucketSize:        20, // MAGIC
+		BootstrapPeers:    DefaultBootstrapPeers(),
 		ProtocolID:        ProtocolIPFS,
 		RoutingTable:      nil,                  // nil because a routing table requires information about the local node. triert.TrieRT will be used if this field is nil.
 		Backends:          map[string]Backend{}, // if empty and [ProtocolIPFS] is used, it'll be populated with the ipns, pk and providers backends
@@ -238,6 +250,14 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("invalid kademlia configuration: %w", err)
 	}
 
+	if c.BucketSize == 0 {
+		return fmt.Errorf("bucket size must not be 0")
+	}
+
+	if len(c.BootstrapPeers) == 0 {
+		return fmt.Errorf("no bootstrap peer")
+	}
+
 	if c.ProtocolID == "" {
 		return fmt.Errorf("protocolID must not be empty")
 	}

v2/config_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -100,4 +101,16 @@ func TestConfig_Validate(t *testing.T) {
 		cfg.Clock = nil
 		assert.Error(t, cfg.Validate())
 	})
+
+	t.Run("zero bucket size", func(t *testing.T) {
+		cfg := DefaultConfig()
+		cfg.BucketSize = 0
+		assert.Error(t, cfg.Validate())
+	})
+
+	t.Run("empty bootstrap peers", func(t *testing.T) {
+		cfg := DefaultConfig()
+		cfg.BootstrapPeers = []peer.AddrInfo{}
+		assert.Error(t, cfg.Validate())
+	})
 }

v2/coord/coordinator.go

@@ -33,6 +33,11 @@
 	// cancel is used to cancel all running goroutines when the coordinator is cleaning up
 	cancel context.CancelFunc
 
+	// done will be closed when the coordinator's eventLoop exits. Block-read
+	// from this channel to wait until resources of this coordinator were
+	// cleaned up
+	done chan struct{}
+
 	// cfg is a copy of the optional configuration supplied to the dht
 	cfg CoordinatorConfig
 
@@ -162,7 +167,7 @@
 
 func NewCoordinator(self kadt.PeerID, rtr 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
 	}
@@ -170,8 +175,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
@@ -180,7 +185,7 @@
 	qpCfg.QueryConcurrency = cfg.RequestConcurrency
 	qpCfg.RequestTimeout = cfg.RequestTimeout
 
-	qp, err := query.NewPool[kadt.Key](kadt.PeerID(self), qpCfg)
+	qp, err := query.NewPool[kadt.Key](self, qpCfg)
 	if err != nil {
 		return nil, fmt.Errorf("query pool: %w", err)
 	}
@@ -235,11 +240,13 @@
 		rtr:    rtr,
 		rt:     rt,
 		cancel: cancel,
+		done:   make(chan struct{}),
 
 		networkBehaviour: networkBehaviour,
 		routingBehaviour: routingBehaviour,
 		queryBehaviour:   queryBehaviour,
 	}
+
 	go d.eventLoop(ctx)
 
 	return d, nil
@@ -248,14 +255,17 @@
 // Close cleans up all resources associated with this Coordinator.
 func (c *Coordinator) Close() error {
 	c.cancel()
+	<-c.done
 	return nil
 }
 
 func (c *Coordinator) ID() kadt.PeerID {
 	return c.self
 }
 
 func (c *Coordinator) eventLoop(ctx context.Context) {
+	defer close(c.done)
+
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.eventLoop")
 	defer span.End()
 	for {
@@ -331,7 +341,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")
 }
 
@@ -446,6 +456,7 @@
 func (c *Coordinator) Bootstrap(ctx context.Context, seeds []kadt.PeerID) error {
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.Bootstrap")
 	defer span.End()
+
 	c.routingBehaviour.Notify(ctx, &EventStartBootstrap{
 		SeedNodes: seeds,
 	})
@@ -468,15 +479,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
 }
 
 // A BufferedRoutingNotifier is a [RoutingNotifier] that buffers [RoutingNotification] events and provides methods
@@ -519,8 +530,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:
 		}
 	}
@@ -545,8 +556,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:
 		}
 	}
@@ -571,8 +582,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/coord/query/query.go

@@ -9,6 +9,7 @@
 	"github.com/plprobelab/go-kademlia/kad"
 	"github.com/plprobelab/go-kademlia/kaderr"
 	"github.com/plprobelab/go-kademlia/key"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/libp2p/go-libp2p-kad-dht/v2/tele"
 )
@@ -93,15 +94,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,
@@ -118,9 +119,13 @@
 	}, nil
 }
 
-func (q *Query[K, N]) Advance(ctx context.Context, ev QueryEvent) QueryState {
-	ctx, span := tele.StartSpan(ctx, "Query.Advance")
-	defer span.End()
+func (q *Query[K, N]) Advance(ctx context.Context, ev QueryEvent) (out QueryState) {
+	ctx, span := tele.StartSpan(ctx, "Query.Advance", trace.WithAttributes(tele.AttrInEvent(ev)))
+	defer func() {
+		span.SetAttributes(tele.AttrOutEvent(out))
+		span.End()
+	}()
+
 	if q.finished {
 		return &StateQueryFinished{
 			QueryID: q.id,
@@ -130,23 +135,20 @@
 
 	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
@@ -174,7 +176,6 @@
 				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,
@@ -191,7 +192,6 @@
 			// 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,
@@ -208,28 +208,27 @@
 				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,
 					Stats:   q.stats,
 					Target:  q.target,
 				}
 				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
@@ -241,7 +240,6 @@
 
 	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,
@@ -251,7 +249,6 @@
 	// 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,
@@ -276,20 +273,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
@@ -310,27 +307,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{}
@@ -367,10 +364,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()
@@ -392,6 +389,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

@@ -59,8 +59,9 @@
 
 // notify must only be called while r.pendingMu is held
 func (r *RoutingBehaviour) notify(ctx context.Context, ev BehaviourEvent) {
-	ctx, span := r.tracer.Start(ctx, "RoutingBehaviour.notify")
+	ctx, span := r.tracer.Start(ctx, "RoutingBehaviour.notify", trace.WithAttributes(attribute.String("event", fmt.Sprintf("%T", ev))))
 	defer span.End()
+
 	switch ev := ev.(type) {
 	case *EventStartBootstrap:
 		span.SetAttributes(attribute.String("event", "EventStartBootstrap"))
@@ -128,10 +129,10 @@
 					NodeID: ev.To,
 				}
 			} else {
 				cmd = &routing.EventIncludeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 					NodeID: ev.To,
 					Error:  fmt.Errorf("response did not include any closer nodes"),
 				}
 			}
 			// attempt to advance the include
 			next, ok := r.advanceInclude(ctx, cmd)
@@ -140,17 +141,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: ev.To,
 				}
 			} else {
 				cmd = &routing.EventProbeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 					NodeID: ev.To,
 					Error:  fmt.Errorf("response did not include any closer nodes"),
 				}
 			}
 			// attempt to advance the probe state machine
 			next, ok := r.advanceProbe(ctx, cmd)
@@ -186,9 +187,9 @@
 				r.pending = append(r.pending, next)
 			}
 		case "probe":
 			cmd := &routing.EventProbeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 				NodeID: ev.To,
 				Error:  ev.Err,
 			}
 			// attempt to advance the probe state machine
 			next, ok := r.advanceProbe(ctx, cmd)
@@ -203,7 +204,7 @@
 		span.SetAttributes(attribute.String("event", "EventNotifyConnectivity"), attribute.String("nodeid", ev.NodeID.String()))
 		// ignore self
 		if r.self.Equal(ev.NodeID) {
 			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]{
@@ -220,8 +221,8 @@
 		}
 		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()))
 
@@ -231,8 +232,8 @@
 		}
 		nextProbe, ok := r.advanceProbe(ctx, cmdProbe)
 		if ok {
 			r.pending = append(r.pending, nextProbe)
 		}
 
 	default:
 		panic(fmt.Sprintf("unexpected dht event: %T", ev))

v2/coord/routing/bootstrap.go

@@ -9,6 +9,7 @@
 	"github.com/plprobelab/go-kademlia/kad"
 	"github.com/plprobelab/go-kademlia/kaderr"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/libp2p/go-libp2p-kad-dht/v2/coord/query"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/tele"
@@ -79,10 +80,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,
@@ -92,13 +93,11 @@
 
 // Advance advances the state of the bootstrap by attempting to advance its query if running.
 func (b *Bootstrap[K, N]) Advance(ctx context.Context, ev BootstrapEvent) BootstrapState {
-	ctx, span := tele.StartSpan(ctx, "Bootstrap.Advance")
+	ctx, span := tele.StartSpan(ctx, "Bootstrap.Advance", trace.WithAttributes(tele.AttrInEvent(ev)))
 	defer span.End()
 
 	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())
 
@@ -111,31 +110,28 @@
 
 		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
+		// ignore, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected event: %T", tev))
 	}
 
 	if b.qry != nil {
@@ -166,11 +162,11 @@
 	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,
@@ -178,17 +174,17 @@
 	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))
 	}
 }
 
@@ -224,11 +220,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 {
@@ -256,7 +252,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()  {}

v2/coord/routing/include.go

@@ -6,10 +6,10 @@
 	"time"
 
 	"github.com/benbjohnson/clock"
-
 	"github.com/plprobelab/go-kademlia/kad"
 	"github.com/plprobelab/go-kademlia/kaderr"
 	"github.com/plprobelab/go-kademlia/key"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/libp2p/go-libp2p-kad-dht/v2/tele"
 )
@@ -85,10 +85,10 @@
 
 func NewInclude[K kad.Key[K], N kad.NodeID[K]](rt kad.RoutingTable[K, N], cfg *IncludeConfig) (*Include[K, N], error) {
 	if cfg == nil {
 		cfg = DefaultIncludeConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	return &Include[K, N]{
 		candidates: newNodeQueue[K, N](cfg.QueueCapacity),
@@ -99,14 +99,15 @@
 }
 
 // Advance advances the state of the include state machine by attempting to advance its query if running.
-func (b *Include[K, N]) Advance(ctx context.Context, ev IncludeEvent) IncludeState {
-	ctx, span := tele.StartSpan(ctx, "Include.Advance")
-	defer span.End()
+func (b *Include[K, N]) Advance(ctx context.Context, ev IncludeEvent) (out IncludeState) {
+	ctx, span := tele.StartSpan(ctx, "Include.Advance", trace.WithAttributes(tele.AttrInEvent(ev)))
+	defer func() {
+		span.SetAttributes(tele.AttrOutEvent(out))
+		span.End()
+	}()
 
 	switch tev := ev.(type) {
-
 	case *EventIncludeAddCandidate[K, N]:
-		span.SetAttributes(tele.AttrEvent("EventIncludeAddCandidate"))
 		// Ignore if already running a check
 		_, checking := b.checks[key.HexString(tev.NodeID.Key())]
 		if checking {
@@ -125,27 +126,23 @@
 		b.candidates.Enqueue(ctx, tev.NodeID)
 
 	case *EventIncludeConnectivityCheckSuccess[K, N]:
-		span.SetAttributes(tele.AttrEvent("EventIncludeConnectivityCheckSuccess"))
 		ch, ok := b.checks[key.HexString(tev.NodeID.Key())]
 		if ok {
 			delete(b.checks, key.HexString(tev.NodeID.Key()))
 			if b.rt.AddNode(tev.NodeID) {
-				span.SetAttributes(tele.AttrOutEvent("StateIncludeRoutingUpdated"))
 				return &StateIncludeRoutingUpdated[K, N]{
 					NodeID: ch.NodeID,
 				}
 			}
 		}
 	case *EventIncludeConnectivityCheckFailure[K, N]:
-		span.SetAttributes(tele.AttrEvent("EventIncludeConnectivityCheckFailure"))
 		span.RecordError(tev.Error)
 		delete(b.checks, key.HexString(tev.NodeID.Key()))
 
 	case *EventIncludePoll:
-		span.SetAttributes(tele.AttrEvent("EventIncludePoll"))
 		// ignore, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected event: %T", tev))
 	}
 
 	if len(b.checks) == b.cfg.Concurrency {
@@ -159,7 +156,6 @@
 	if !ok {
 		// No candidate in queue
 		if len(b.checks) > 0 {
-			span.SetAttributes(tele.AttrOutEvent("StateIncludeWaitingWithCapacity"))
 			return &StateIncludeWaitingWithCapacity{}
 		}
 		return &StateIncludeIdle{}
@@ -171,7 +167,6 @@
 	}
 
 	// Ask the node to find itself
-	span.SetAttributes(tele.AttrOutEvent("StateIncludeConnectivityCheck"))
 	return &StateIncludeConnectivityCheck[K, N]{
 		NodeID: candidate,
 	}
@@ -196,12 +191,12 @@
 // added and false otherwise.
 func (q *nodeQueue[K, N]) Enqueue(ctx context.Context, id N) bool {
 	if len(q.nodes) == q.capacity {
 		return false
 	}
 
 	if _, exists := q.keys[key.HexString(id.Key())]; exists {
 		return false
 	}
 
 	q.nodes = append(q.nodes, id)
 	q.keys[key.HexString(id.Key())] = struct{}{}
@@ -259,12 +254,12 @@
 }
 
 // includeState() ensures that only Include states can be assigned to an IncludeState.
 func (*StateIncludeConnectivityCheck[K, N]) includeState() {}
 func (*StateIncludeIdle) includeState()                    {}
 func (*StateIncludeWaitingAtCapacity) includeState()       {}
 func (*StateIncludeWaitingWithCapacity) includeState()     {}
 func (*StateIncludeWaitingFull) includeState()             {}
 func (*StateIncludeRoutingUpdated[K, N]) includeState()    {}
 
 // IncludeEvent is an event intended to advance the state of an [Include].
 type IncludeEvent interface {
@@ -291,7 +286,7 @@
 }
 
 // includeEvent() ensures that only events accepted by an [Include] can be assigned to the [IncludeEvent] interface.
 func (*EventIncludePoll) includeEvent()                           {}
 func (*EventIncludeAddCandidate[K, N]) includeEvent()             {}
 func (*EventIncludeConnectivityCheckSuccess[K, N]) includeEvent() {}
 func (*EventIncludeConnectivityCheckFailure[K, N]) includeEvent() {}