Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(wakuv2): logout / login delay #4336

Merged
merged 4 commits into from
Nov 17, 2023
Merged

fix(wakuv2): logout / login delay #4336

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6c1a773
fix(waku2): use a cancelable context for initial bootnode discovery
richard-ramos Nov 17, 2023
7a792c2
fix: start idService and add wg.Done to fnApply
richard-ramos Nov 17, 2023
76f55bc
fix: memory aliasing in loop
richard-ramos Nov 17, 2023
4f1c4e4
fix: reset context on stop
richard-ramos Nov 17, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion protocol/transport/filters_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,8 @@ func (f *FiltersManager) InitCommunityFilters(communityFiltersToInitialize []Com
// TODO: requests to join / cancels are currently being sent into the default waku topic.
// They must be sent into an specific non protected shard
for _, pubsubTopic := range topics {
identityStr := PublicKeyToStr(&cf.PrivKey.PublicKey)
pk := &cf.PrivKey.PublicKey
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So here, given that we're taking an address of loop variable cf, this pk param would always point to the same var, correct?

Copy link
Member Author

@richard-ramos richard-ramos Nov 17, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah!
Earlier I updated golangci-lint on my machine and when I executed make lint saw a warning!

identityStr := PublicKeyToStr(pk)
rawFilter, err := f.addAsymmetric(identityStr, pubsubTopic, cf.PrivKey, true)
if err != nil {
f.logger.Debug("could not register community filter", zap.Error(err))
Expand Down
31 changes: 20 additions & 11 deletions wakuv2/waku.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,6 +205,8 @@ func New(nodeKey string, fleet string, cfg *Config, logger *zap.Logger, appDB *s

logger.Info("starting wakuv2 with config", zap.Any("config", cfg))

ctx, cancel := context.WithCancel(context.Background())

waku := &Waku{
appDB: appDB,
cfg: cfg,
Expand All @@ -216,6 +218,8 @@ func New(nodeKey string, fleet string, cfg *Config, logger *zap.Logger, appDB *s
sendQueue: make(chan *protocol.Envelope, 1000),
connStatusChan: make(chan node.ConnStatus, 100),
connStatusSubscriptions: make(map[string]*types.ConnStatusSubscription),
ctx: ctx,
cancel: cancel,
wg: sync.WaitGroup{},
dnsAddressCache: make(map[string][]dnsdisc.DiscoveredNode),
dnsAddressCacheLock: &sync.RWMutex{},
Expand Down Expand Up @@ -284,7 +288,7 @@ func New(nodeKey string, fleet string, cfg *Config, logger *zap.Logger, appDB *s
}

if cfg.EnableDiscV5 {
bootnodes, err := waku.getDiscV5BootstrapNodes(context.Background(), cfg.DiscV5BootstrapNodes)
bootnodes, err := waku.getDiscV5BootstrapNodes(waku.ctx, cfg.DiscV5BootstrapNodes)
if err != nil {
logger.Error("failed to get bootstrap nodes", zap.Error(err))
return nil, err
Expand Down Expand Up @@ -422,6 +426,7 @@ func (w *Waku) dnsDiscover(ctx context.Context, enrtreeAddress string, apply fnA

func (w *Waku) addWakuV2Peers(ctx context.Context, cfg *Config) error {
fnApply := func(d dnsdisc.DiscoveredNode, wg *sync.WaitGroup) {
defer wg.Done()
if len(d.PeerInfo.Addrs) != 0 {
go w.identifyAndConnect(ctx, w.settings.LightClient, d.PeerInfo)
}
Expand Down Expand Up @@ -1018,10 +1023,10 @@ func (w *Waku) broadcast() {
logger := w.logger.With(zap.String("envelopeHash", hexutil.Encode(envelope.Hash())), zap.String("pubsubTopic", pubsubTopic), zap.String("contentTopic", envelope.Message().ContentTopic), zap.Int64("timestamp", envelope.Message().Timestamp))
if w.settings.LightClient {
logger.Info("publishing message via lightpush")
_, err = w.node.Lightpush().Publish(context.Background(), envelope.Message(), lightpush.WithPubSubTopic(pubsubTopic))
_, err = w.node.Lightpush().Publish(w.ctx, envelope.Message(), lightpush.WithPubSubTopic(pubsubTopic))
} else {
logger.Info("publishing message via relay")
_, err = w.node.Relay().Publish(context.Background(), envelope.Message(), relay.WithPubSubTopic(pubsubTopic))
_, err = w.node.Relay().Publish(w.ctx, envelope.Message(), relay.WithPubSubTopic(pubsubTopic))
}

if err != nil {
Expand Down Expand Up @@ -1137,6 +1142,9 @@ func (w *Waku) Query(ctx context.Context, peerID peer.ID, pubsubTopic string, to
// Start implements node.Service, starting the background data propagation thread
// of the Waku protocol.
func (w *Waku) Start() error {
if w.ctx == nil {
w.ctx, w.cancel = context.WithCancel(context.Background())
}

var err error
if w.node, err = node.New(w.settings.Options...); err != nil {
Expand All @@ -1145,11 +1153,7 @@ func (w *Waku) Start() error {

w.connectionChanged = make(chan struct{})

ctx, cancel := context.WithCancel(context.Background())
w.ctx = ctx
w.cancel = cancel

if err = w.node.Start(ctx); err != nil {
if err = w.node.Start(w.ctx); err != nil {
return fmt.Errorf("failed to start go-waku node: %v", err)
}

Expand All @@ -1159,15 +1163,16 @@ func (w *Waku) Start() error {
if err != nil {
return err
}
idService.Start()

w.identifyService = idService

if err = w.addWakuV2Peers(ctx, w.cfg); err != nil {
if err = w.addWakuV2Peers(w.ctx, w.cfg); err != nil {
return fmt.Errorf("failed to add wakuv2 peers: %v", err)
}

if w.cfg.EnableDiscV5 {
err := w.node.DiscV5().Start(ctx)
err := w.node.DiscV5().Start(w.ctx)
if err != nil {
return err
}
Expand Down Expand Up @@ -1209,7 +1214,7 @@ func (w *Waku) Start() error {
w.logger.Info("Restarting DiscV5: online and is not connected")
isConnected = true
if w.node.DiscV5().ErrOnNotRunning() != nil {
err := w.node.DiscV5().Start(ctx)
err := w.node.DiscV5().Start(w.ctx)
if err != nil {
w.logger.Error("Could not start DiscV5", zap.Error(err))
}
Expand Down Expand Up @@ -1301,6 +1306,10 @@ func (w *Waku) Stop() error {

close(w.connectionChanged)
w.wg.Wait()

w.ctx = nil
w.cancel = nil

return nil
}

Expand Down