Avoid BlossomSubRouter race condition (#364)

node/p2p/blossomsub.go

@@ -448,6 +448,9 @@ func NewBlossomSub(
 		internal.PeerAddrInfosToPeerIDSlice(bootstrappers),
 		true,
 	)))
+	blossomOpts = append(blossomOpts, blossomsub.WithDiscovery(
+		internal.NewPeerConnectorDiscovery(discovery),
+	))
 
 	params := toBlossomSubParams(p2pConfig)
 	rt := blossomsub.NewBlossomSubRouter(h, params, bs.network)
@@ -463,18 +466,6 @@ func NewBlossomSub(
 	bs.h = h
 	bs.signKey = privKey
 
-	go func() {
-		for {
-			time.Sleep(30 * time.Second)
-			for _, mask := range pubsub.GetBitmasks() {
-				if !rt.EnoughPeers([]byte(mask), 0) {
-					_ = discovery.Connect(ctx)
-					break
-				}
-			}
-		}
-	}()
-
 	return bs
 }
 

node/p2p/internal/discovery.go

@@ -0,0 +1,37 @@
+package internal
+
+import (
+	"context"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/discovery"
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+type peerConnectorDiscovery struct {
+	connector PeerConnector
+}
+
+var _ discovery.Discovery = (*peerConnectorDiscovery)(nil)
+
+// Advertise implements discovery.Advertiser.
+func (d *peerConnectorDiscovery) Advertise(ctx context.Context, ns string, opts ...discovery.Option) (time.Duration, error) {
+	return time.Duration(1<<63 - 1), nil
+}
+
+// FindPeers implements discovery.Discoverer.
+func (d *peerConnectorDiscovery) FindPeers(ctx context.Context, ns string, opts ...discovery.Option) (<-chan peer.AddrInfo, error) {
+	if err := d.connector.Connect(ctx); err != nil {
+		return nil, err
+	}
+	ch := make(chan peer.AddrInfo)
+	close(ch)
+	return ch, nil
+}
+
+// NewPeerConnectorDiscovery creates a new peer connector discovery.
+// The discovery instance does not do any advertisements and just triggers
+// the peer connector once FindPeers is called.
+func NewPeerConnectorDiscovery(connector PeerConnector) discovery.Discovery {
+	return &peerConnectorDiscovery{connector: connector}
+}