diff --git a/waku/v2/node/localnode.go b/waku/v2/node/localnode.go index 5b042734..c3e427b7 100644 --- a/waku/v2/node/localnode.go +++ b/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 { diff --git a/waku/v2/node/wakunode2.go b/waku/v2/node/wakunode2.go index abb5ca60..da2de817 100644 --- a/waku/v2/node/wakunode2.go +++ b/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 {