fix: start watchTopicShards once (#1298)

Author: igor-sirotin

waku/v2/node/localnode.go

@@ -348,6 +348,10 @@ func (w *WakuNode) SetRelayShards(rs protocol.RelayShards) error {
 }
 
 func (w *WakuNode) watchTopicShards(ctx context.Context) error {
+	if !w.watchingRelayShards.CompareAndSwap(false, true) {
+		return nil
+	}
+
 	evtRelaySubscribed, err := w.Relay().Events().Subscribe(new(relay.EvtRelaySubscribed))
 	if err != nil {
 		return err
@@ -358,10 +362,13 @@ func (w *WakuNode) watchTopicShards(ctx context.Context) error {
 		return err
 	}
 
+	w.wg.Add(1)
+
 	go func() {
 		defer utils.LogOnPanic()
 		defer evtRelaySubscribed.Close()
 		defer evtRelayUnsubscribed.Close()
+		defer w.wg.Done()
 
 		for {
 			select {

waku/v2/node/wakunode2.go

@@ -5,6 +5,7 @@ import (
 	"math/rand"
 	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	backoffv4 "github.com/cenkalti/backoff/v4"
@@ -122,6 +123,8 @@ type WakuNode struct {
 	storeFactory storeFactory
 
 	peermanager *peermanager.PeerManager
+
+	watchingRelayShards atomic.Bool
 }
 
 func defaultStoreFactory(w *WakuNode) legacy_store.Store {