fix: race condition in peer connector / manager interaction (#837)

waku/v2/peermanager/peer_connector.go

@@ -35,12 +35,17 @@ type PeerConnectionStrategy struct {
 	paused      atomic.Bool
 	dialTimeout time.Duration
 	*CommonDiscoveryService
-	subscriptions []<-chan PeerData
+	subscriptions []subscription
 
 	backoff backoff.BackoffFactory
 	logger  *zap.Logger
 }
 
+type subscription struct {
+	ctx context.Context
+	ch  <-chan PeerData
+}
+
 // backoff describes the strategy used to decide how long to backoff after previously attempting to connect to a peer
 func getBackOff() backoff.BackoffFactory {
 	rngSrc := rand.NewSource(rand.Int63())
@@ -85,32 +90,36 @@ func (c *PeerConnectionStrategy) Subscribe(ctx context.Context, ch <-chan PeerDa
 	// if not running yet, store the subscription and return
 	if err := c.ErrOnNotRunning(); err != nil {
 		c.mux.Lock()
-		c.subscriptions = append(c.subscriptions, ch)
+		c.subscriptions = append(c.subscriptions, subscription{ctx, ch})
 		c.mux.Unlock()
 		return
 	}
 	// if running start a goroutine to consume the subscription
 	c.WaitGroup().Add(1)
 	go func() {
 		defer c.WaitGroup().Done()
-		c.consumeSubscription(ch)
+		c.consumeSubscription(subscription{ctx, ch})
 	}()
 }
 
-func (c *PeerConnectionStrategy) consumeSubscription(ch <-chan PeerData) {
+func (c *PeerConnectionStrategy) consumeSubscription(s subscription) {
 	for {
 		// for returning from the loop when peerConnector is paused.
 		select {
 		case <-c.Context().Done():
 			return
+		case <-s.ctx.Done():
+			return
 		default:
 		}
 		//
 		if !c.isPaused() {
 			select {
 			case <-c.Context().Done():
 				return
-			case p, ok := <-ch:
+			case <-s.ctx.Done():
+				return
+			case p, ok := <-s.ch:
 				if !ok {
 					return
 				}
@@ -166,7 +175,7 @@ func (c *PeerConnectionStrategy) isPaused() bool {
 func (c *PeerConnectionStrategy) consumeSubscriptions() {
 	for _, subs := range c.subscriptions {
 		c.WaitGroup().Add(1)
-		go func(s <-chan PeerData) {
+		go func(s subscription) {
 			defer c.WaitGroup().Done()
 			c.consumeSubscription(s)
 		}(subs)

waku/v2/peermanager/peer_manager.go

@@ -339,7 +339,7 @@ func (pm *PeerManager) AddDiscoveredPeer(p PeerData, connectNow bool) {
 		}
 	}
 	if connectNow {
-		pm.peerConnector.PushToChan(p)
+		go pm.peerConnector.PushToChan(p)
 	}
 }