fix: flaky test by skipping connectivity checks

v2/config.go

@@ -395,16 +395,21 @@ type QueryConfig struct {
 	// operation. A DefaultQuorum of 0 means that we search the network until
 	// we have exhausted the keyspace.
 	DefaultQuorum int
+
+	// SkipConnectivityCheck defines whether we do a connectivity check before
+	// we add peers to the routing table.
+	SkipConnectivityCheck bool
 }
 
 // DefaultQueryConfig returns the default query configuration options for a DHT.
 func DefaultQueryConfig() *QueryConfig {
 	return &QueryConfig{
-		Concurrency:        3,               // MAGIC
-		Timeout:            5 * time.Minute, // MAGIC
-		RequestConcurrency: 3,               // MAGIC
-		RequestTimeout:     time.Minute,     // MAGIC
-		DefaultQuorum:      0,               // MAGIC
+		Concurrency:           3,               // MAGIC
+		Timeout:               5 * time.Minute, // MAGIC
+		RequestConcurrency:    3,               // MAGIC
+		RequestTimeout:        time.Minute,     // MAGIC
+		DefaultQuorum:         0,               // MAGIC
+		SkipConnectivityCheck: false,
 	}
 }
 

v2/dht.go

@@ -100,7 +100,7 @@
 	} else if cfg.ProtocolID == ProtocolIPFS {
 		d.backends, err = d.initAminoBackends()
 		if err != nil {
 			return nil, fmt.Errorf("init amino backends: %w", err)
 		}
 	}
 
@@ -128,6 +128,7 @@
 	coordCfg.Routing.Logger = cfg.Logger.With("behaviour", "routing")
 	coordCfg.Routing.Tracer = cfg.TracerProvider.Tracer(tele.TracerName)
 	coordCfg.Routing.Meter = cfg.MeterProvider.Meter(tele.MeterName)
+	coordCfg.Routing.IncludeSkipCheck = cfg.Query.SkipConnectivityCheck
 
 	rtr := &router{
 		host:       h,
@@ -172,18 +173,18 @@
 	)
 
 	if d.cfg.Datastore != nil {
 		dstore = d.cfg.Datastore
 	} else if dstore, err = InMemoryDatastore(); err != nil {
 		return nil, fmt.Errorf("new default datastore: %w", err)
 	}
 
 	// wrap datastore in open telemetry tracing
 	dstore = trace.New(dstore, d.tele.Tracer)
 
 	pbeCfg, err := DefaultProviderBackendConfig()
 	if err != nil {
 		return nil, fmt.Errorf("default provider config: %w", err)
 	}
 	pbeCfg.Logger = d.cfg.Logger
 	pbeCfg.AddressFilter = d.cfg.AddressFilter
 	pbeCfg.Tele = d.tele
@@ -191,26 +192,26 @@
 
 	pbe, err := NewBackendProvider(d.host.Peerstore(), dstore, pbeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("new provider backend: %w", err)
 	}
 
 	rbeCfg, err := DefaultRecordBackendConfig()
 	if err != nil {
 		return nil, fmt.Errorf("default provider config: %w", err)
 	}
 	rbeCfg.Logger = d.cfg.Logger
 	rbeCfg.Tele = d.tele
 	rbeCfg.clk = d.cfg.Clock
 
 	ipnsBe, err := NewBackendIPNS(dstore, d.host.Peerstore(), rbeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("new ipns backend: %w", err)
 	}
 
 	pkBe, err := NewBackendPublicKey(dstore, rbeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("new public key backend: %w", err)
 	}
 
 	return map[string]Backend{
 		namespaceIPNS:      ipnsBe,
@@ -226,11 +227,11 @@
 	}
 
 	if err := d.sub.Close(); err != nil {
 		d.debugErr(err, "failed closing event bus subscription")
 	}
 
 	if err := d.kad.Close(); err != nil {
 		d.debugErr(err, "failed closing coordinator")
 	}
 
 	for ns, b := range d.backends {
@@ -240,7 +241,7 @@
 		}
 
 		if err := closer.Close(); err != nil {
 			d.warnErr(err, "failed closing backend", "namespace", ns)
 		}
 	}
 
@@ -254,7 +255,7 @@
 	if d.cfg.ProtocolID == ProtocolIPFS && d.cfg.Datastore == nil {
 		if pbe, err := typedBackend[*ProvidersBackend](d, namespaceProviders); err == nil {
 			if err := pbe.datastore.Close(); err != nil {
 				d.warnErr(err, "failed closing in memory datastore")
 			}
 		}
 	}
@@ -268,7 +269,7 @@
 			}
 
 			if err := s.Reset(); err != nil {
 				d.debugErr(err, "failed closing stream")
 			}
 		}
 	}
@@ -326,7 +327,7 @@
 			}
 
 			if err := s.Reset(); err != nil {
 				d.debugErr(err, "failed closing stream")
 			}
 		}
 	}
@@ -345,22 +346,22 @@
 		d.log.Warn(msg, tele.LogAttrError(err))
 		return
 	}
 	d.log.With(args...).Warn(msg, tele.LogAttrError(err))
 }
 
 // debugErr is a helper method that uses the slogger of the DHT and writes a
 // debug log line with the given message alongside the error. args is a list of
 // key/value pairs or slog.Attrs that will be included with the log message. If the error
 // is nil, this method is a no-op.
 func (d *DHT) debugErr(err error, msg string, args ...any) {
 	if err == nil {
 		return
 	}
 	if len(args) == 0 {
 		d.log.Debug(msg, tele.LogAttrError(err))
 		return
 	}
 	d.log.With(args...).Debug(msg, tele.LogAttrError(err))
 }
 
 // AddAddresses suggests peers and their associated addresses to be added to the routing table.

v2/internal/coord/routing.go

@@ -62,6 +62,9 @@
 	// ProbeCheckInterval is the time interval the behaviour should use between connectivity checks for the same node in the routing table.
 	ProbeCheckInterval time.Duration
 
+	// IncludeSkipCheck indicates whether we perform connectivity checks before we add a peer to the routing table.
+	IncludeSkipCheck bool
+
 	// IncludeQueueCapacity is the maximum number of nodes the behaviour should keep queued as candidates for inclusion in the routing table.
 	IncludeQueueCapacity int
 
@@ -268,6 +271,7 @@
 		ProbeRequestConcurrency: 3,             // MAGIC
 		ProbeCheckInterval:      6 * time.Hour, // MAGIC
 
+		IncludeSkipCheck:          false,
 		IncludeRequestConcurrency: 3,   // MAGIC
 		IncludeQueueCapacity:      128, // MAGIC
 
@@ -307,12 +311,28 @@
 	ready     chan struct{}
 }
 
+type Recording2SM[E any, S any] struct {
+	State    S
+	Received E
+}
+
+func NewRecording2SM[E any, S any](response S) *Recording2SM[E, S] {
+	return &Recording2SM[E, S]{
+		State: response,
+	}
+}
+
+func (r *Recording2SM[E, S]) Advance(ctx context.Context, e E) S {
+	r.Received = e
+	return r.State
+}
+
 func NewRoutingBehaviour(self kadt.PeerID, rt routing.RoutingTableCpl[kadt.Key, kadt.PeerID], cfg *RoutingConfig) (*RoutingBehaviour, error) {
 	if cfg == nil {
 		cfg = DefaultRoutingConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	bootstrapCfg := routing.DefaultBootstrapConfig()
 	bootstrapCfg.Clock = cfg.Clock
@@ -324,21 +344,22 @@
 
 	bootstrap, err := routing.NewBootstrap[kadt.Key](self, bootstrapCfg)
 	if err != nil {
 		return nil, fmt.Errorf("bootstrap: %w", err)
 	}
 
 	includeCfg := routing.DefaultIncludeConfig()
 	includeCfg.Clock = cfg.Clock
 	includeCfg.Tracer = cfg.Tracer
 	includeCfg.Meter = cfg.Meter
+	includeCfg.SkipCheck = cfg.IncludeSkipCheck
 	includeCfg.Timeout = cfg.ConnectivityCheckTimeout
 	includeCfg.QueueCapacity = cfg.IncludeQueueCapacity
 	includeCfg.Concurrency = cfg.IncludeRequestConcurrency
 
 	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
@@ -350,8 +371,8 @@
 
 	probe, err := routing.NewProbe[kadt.Key](rt, probeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("probe: %w", err)
 	}
 
 	exploreCfg := routing.DefaultExploreConfig()
 	exploreCfg.Clock = cfg.Clock
@@ -363,13 +384,13 @@
 
 	schedule, err := routing.NewDynamicExploreSchedule(cfg.ExploreMaximumCpl, cfg.Clock.Now(), cfg.ExploreInterval, cfg.ExploreIntervalMultiplier, cfg.ExploreIntervalJitter)
 	if err != nil {
 		return nil, fmt.Errorf("explore schedule: %w", err)
 	}
 
 	explore, err := routing.NewExplore[kadt.Key](self, rt, cplutil.GenRandPeerID, schedule, exploreCfg)
 	if err != nil {
 		return nil, fmt.Errorf("explore: %w", err)
 	}
 
 	return ComposeRoutingBehaviour(self, bootstrap, include, probe, explore, cfg)
 }
@@ -385,10 +406,10 @@
 	cfg *RoutingConfig,
 ) (*RoutingBehaviour, error) {
 	if cfg == nil {
 		cfg = DefaultRoutingConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	r := &RoutingBehaviour{
 		self:      self,
@@ -452,8 +473,8 @@
 		// attempt to advance the probe state machine
 		next, ok := r.advanceProbe(ctx, cmd)
 		if ok {
 			r.pending = append(r.pending, next)
 		}
 
 	case *EventGetCloserNodesSuccess:
 		span.SetAttributes(attribute.String("event", "EventGetCloserNodesSuccess"), attribute.String("queryid", string(ev.QueryID)), attribute.String("nodeid", ev.To.String()))
@@ -483,35 +504,35 @@
 					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)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		case ProbeQueryID:
 			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)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		case routing.ExploreQueryID:
 			for _, info := range ev.CloserNodes {
@@ -525,11 +546,11 @@
 			}
 			next, ok := r.advanceExplore(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		default:
 			panic(fmt.Sprintf("unexpected query id: %s", ev.QueryID))
 		}
 	case *EventGetCloserNodesFailure:
 		span.SetAttributes(attribute.String("event", "EventGetCloserNodesFailure"), attribute.String("queryid", string(ev.QueryID)), attribute.String("nodeid", ev.To.String()))
@@ -543,8 +564,8 @@
 			// attempt to advance the bootstrap
 			next, ok := r.advanceBootstrap(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 		case IncludeQueryID:
 			cmd := &routing.EventIncludeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 				NodeID: ev.To,
@@ -553,18 +574,18 @@
 			// attempt to advance the include state machine
 			next, ok := r.advanceInclude(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 		case ProbeQueryID:
 			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)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 		case routing.ExploreQueryID:
 			cmd := &routing.EventExploreFindCloserFailure[kadt.Key, kadt.PeerID]{
 				NodeID: ev.To,
@@ -573,17 +594,17 @@
 			// attempt to advance the explore
 			next, ok := r.advanceExplore(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		default:
 			panic(fmt.Sprintf("unexpected query id: %s", ev.QueryID))
 		}
 	case *EventNotifyConnectivity:
 		span.SetAttributes(attribute.String("event", "EventNotifyConnectivity"), attribute.String("nodeid", ev.NodeID.String()))
 		// ignore self
 		if r.self.Equal(ev.NodeID) {
 			break
 		}
 		r.cfg.Logger.Debug("peer has connectivity", tele.LogAttrPeerID(ev.NodeID))
 
@@ -602,8 +623,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()))
 
@@ -613,13 +634,13 @@
 		}
 		nextProbe, ok := r.advanceProbe(ctx, cmdProbe)
 		if ok {
 			r.pending = append(r.pending, nextProbe)
 		}
 	case *EventRoutingPoll:
 		r.pollChildren(ctx)
 
 	default:
 		panic(fmt.Sprintf("unexpected dht event: %T", ev))
 	}
 
 	if len(r.pending) > 0 {
@@ -671,13 +692,13 @@
 func (r *RoutingBehaviour) pollChildren(ctx context.Context) {
 	ev, ok := r.advanceBootstrap(ctx, &routing.EventBootstrapPoll{})
 	if ok {
 		r.pending = append(r.pending, ev)
 	}
 
 	ev, ok = r.advanceInclude(ctx, &routing.EventIncludePoll{})
 	if ok {
 		r.pending = append(r.pending, ev)
 	}
 
 	ev, ok = r.advanceProbe(ctx, &routing.EventProbePoll{})
 	if ok {
@@ -704,7 +725,7 @@
 			Notify:  r,
 		}, true
 
 	case *routing.StateBootstrapWaiting:
 		// bootstrap waiting for a message response, nothing to do
 	case *routing.StateBootstrapFinished:
 		r.cfg.Logger.Debug("bootstrap finished", slog.Duration("elapsed", st.Stats.End.Sub(st.Stats.Start)), slog.Int("requests", st.Stats.Requests), slog.Int("failures", st.Stats.Failure))
@@ -713,8 +734,8 @@
 		}, true
 	case *routing.StateBootstrapIdle:
 		// bootstrap not running, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected bootstrap state: %T", st))
 	}
 
 	return nil, false
@@ -751,16 +772,16 @@
 		return &EventRoutingUpdated{
 			NodeID: st.NodeID,
 		}, true
 	case *routing.StateIncludeWaitingAtCapacity:
 		// nothing to do except wait for message response or timeout
 	case *routing.StateIncludeWaitingWithCapacity:
 		// nothing to do except wait for message response or timeout
 	case *routing.StateIncludeWaitingFull:
 		// nothing to do except wait for message response or timeout
 	case *routing.StateIncludeIdle:
 		// nothing to do except wait for new nodes to be added to queue
 	default:
 		panic(fmt.Sprintf("unexpected include state: %T", st))
 	}
 
 	return nil, false
@@ -793,17 +814,17 @@
 		r.notify(ctx, &EventAddNode{
 			NodeID: st.NodeID,
 		})
 	case *routing.StateProbeWaitingAtCapacity:
 		// the probe state machine is waiting for responses for checks and the maximum number of concurrent checks has been reached.
 		// nothing to do except wait for message response or timeout
 	case *routing.StateProbeWaitingWithCapacity:
 		// the probe state machine is waiting for responses for checks but has capacity to perform more
 		// nothing to do except wait for message response or timeout
 	case *routing.StateProbeIdle:
 		// the probe state machine is not running any checks.
 		// nothing to do except wait for message response or timeout
 	default:
 		panic(fmt.Sprintf("unexpected include state: %T", st))
 	}
 
 	return nil, false
@@ -824,18 +845,18 @@
 			Notify:  r,
 		}, true
 
 	case *routing.StateExploreWaiting:
 		// explore waiting for a message response, nothing to do
 	case *routing.StateExploreQueryFinished:
 		// nothing to do except notify via telemetry
 	case *routing.StateExploreQueryTimeout:
 		// nothing to do except notify via telemetry
 	case *routing.StateExploreFailure:
 		r.cfg.Logger.Warn("explore failure", slog.Int("cpl", st.Cpl), tele.LogAttrError(st.Error))
 	case *routing.StateExploreIdle:
 		// bootstrap not running, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected explore state: %T", st))
 	}
 
 	return nil, false

v2/internal/coord/routing/include.go

@@ -61,12 +61,9 @@
 	Concurrency   int           // the maximum number of include checks that may be in progress at any one time
 	Timeout       time.Duration // the time to wait before terminating a check that is not making progress
 	Clock         clock.Clock   // a clock that may replaced by a mock when testing
-
-	// Tracer is the tracer that should be used to trace execution.
-	Tracer trace.Tracer
-
-	// Meter is the meter that should be used to record metrics.
-	Meter metric.Meter
+	SkipCheck     bool          // whether to skip connectivity checks and add any node passed to this state machine
+	Tracer        trace.Tracer  // Tracer is the tracer that should be used to trace execution.
+	Meter         metric.Meter  // Meter is the meter that should be used to record metrics.
 }
 
 // Validate checks the configuration options and returns an error if any have invalid values.
@@ -100,18 +97,18 @@
 	}
 
 	if cfg.Tracer == nil {
 		return &errs.ConfigurationError{
 			Component: "IncludeConfig",
 			Err:       fmt.Errorf("tracer must not be nil"),
 		}
 	}
 
 	if cfg.Meter == nil {
 		return &errs.ConfigurationError{
 			Component: "IncludeConfig",
 			Err:       fmt.Errorf("meter must not be nil"),
 		}
 	}
 
 	return nil
 }
@@ -124,6 +121,7 @@
 		Tracer: tele.NoopTracer(),
 		Meter:  tele.NoopMeter(),
 
+		SkipCheck:     false,
 		Concurrency:   3,
 		Timeout:       time.Minute,
 		QueueCapacity: 128,
@@ -132,10 +130,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
 	}
 
 	in := &Include[K, N]{
 		candidates: newNodeQueue[K, N](cfg.QueueCapacity),
@@ -152,8 +150,8 @@
 		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_checks_sent counter: %w", err)
 	}
 
 	in.counterChecksPassed, err = cfg.Meter.Int64Counter(
 		"include_checks_passed",
@@ -161,8 +159,8 @@
 		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_checks_passed counter: %w", err)
 	}
 
 	in.counterChecksFailed, err = cfg.Meter.Int64Counter(
 		"include_checks_failed",
@@ -170,8 +168,8 @@
 		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_checks_failed counter: %w", err)
 	}
 
 	in.counterCandidatesDroppedCapacity, err = cfg.Meter.Int64Counter(
 		"include_candidates_dropped_capacity",
@@ -179,21 +177,21 @@
 		metric.WithUnit("1"),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_candidates_dropped_capacity counter: %w", err)
 	}
 
 	in.gaugeCandidateCount, err = cfg.Meter.Int64ObservableGauge(
 		"include_candidate_count",
 		metric.WithDescription("Total number of nodes in the include state machine's candidate queue"),
 		metric.WithUnit("1"),
 		metric.WithInt64Callback(func(ctx context.Context, o metric.Int64Observer) error {
 			o.Observe(in.candidateCount.Load())
 			return nil
 		}),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create include_candidate_count counter: %w", err)
 	}
 
 	return in, nil
 }
@@ -209,6 +207,15 @@
 
 	switch tev := ev.(type) {
 	case *EventIncludeAddCandidate[K, N]:
+
+		if in.cfg.SkipCheck {
+			if in.rt.AddNode(tev.NodeID) {
+				return &StateIncludeRoutingUpdated[K, N]{NodeID: tev.NodeID}
+			} else {
+				return &StateIncludeIdle{}
+			}
+		}
+
 		// Ignore if already running a check
 		_, checking := in.checks[key.HexString(tev.NodeID.Key())]
 		if checking {
@@ -245,8 +252,8 @@
 
 	case *EventIncludePoll:
 		// ignore, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected event: %T", tev))
 	}
 
 	if len(in.checks) == in.cfg.Concurrency {
@@ -297,12 +304,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{}{}
@@ -364,12 +371,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 {
@@ -396,7 +403,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() {}

v2/routing_test.go

@@ -682,6 +682,8 @@ func (suite *SearchValueQuorumTestSuite) SetupTest() {
 
 	cfg := DefaultConfig()
 	cfg.Clock = clk
+	cfg.Query.SkipConnectivityCheck = true
+
 	top := NewTopology(t)
 
 	// init privileged DHT server