use event bus for monitoring peer connections and protocol updates #536

jefft0 · 2023-07-06T15:41:43Z

This fixes a bug where another peer is not added to a topic after subscribing. This bug happens, for example, when a discovery system (like MDNS) is set up before pubsub initialization. Even though the peers have been discovered, they are not added to the new pubsub topics.

The bug can be seen by running TestNotifyPeerProtocolsUpdated in the master branch:

git clone https://github.com/libp2p/go-libp2p-pubsub
cd go-libp2p-pubsub
curl https://raw.githubusercontent.com/jefft0/go-libp2p-pubsub/fix/notify-protocol-update/notify_test.go -o notify_test.go
go test -run TestNotifyPeerProtocolsUpdated .

It prints "topic1 should at least have 1 peer". In our code we had to workaround this by disabling discovery (MDNS) and deferring discovery until after the pubsub protocols are registered. However, this pull request fixes the problem so that a workaround is not needed. It uses the event bus to monitor for peer connections and protocol updates. When hosts[1] joins topic1, the other peer is automatically added and the test passes.

Signed-off-by: gfanton <8671905+gfanton@users.noreply.github.com>

jefft0 · 2023-07-06T15:50:06Z

Some background if it helps: We discovered and created a fix for this issue while working on this simple example of our networking library that is built on libp2p.
https://wesh.network/posts/share-contact-and-send-message

vyzo

Thank you for this pr.

This is something that we have had to do for a while, so it is a step on the right direction.

At first glance it looks good, but I will do a second more thorough pass to make sure we don't break stuff.

vyzo · 2023-07-06T16:23:04Z

pubsub.go


 	ps.val.Start(ps)

 	go ps.processLoop(ctx)

-	(*PubSubNotif)(ps).Initialize()
+	if err := (*PubSubNotif)(ps).startMonitoring(); err != nil {


Let's do this check before spawning the processLoop, as if it fails the system will not work at all.

Hello @vyzo . With the help of @gfanton , I added a second commit to address this comment. It more closely follows the logic which was there before.

Thanks for the thumbs up. Do we need to address any other issues with this PR?

Signed-off-by: Jeff Thompson <jeff@thefirst.org>

vyzo

This is a good start, left some comments requiring changes.

notify.go

vyzo · 2023-07-27T18:54:35Z

notify.go

-func (p *PubSubNotif) Initialize() {
-	isTransient := func(pid peer.ID) bool {
-		for _, c := range p.host.Network().ConnsToPeer(pid) {
-			if !c.Stat().Transient {
-				return false
-			}


We want to keep this code around; please don't remove it and run it right after initializing the bus and before starting the monitoring goroutine.

So the initialization code is still there (below), but I believe it is racey.

See comments below.

vyzo · 2023-07-27T18:55:42Z

notify.go

+				if evt.Connectedness == network.Connected {
+					go p.AddPeers(evt.Peer)


we should process the disconnect events as well and get rid of the nasty code that handles current peer disconnections.

The latest commit handles the NotConnected event and calls RemovePeers. How does it look?

vyzo · 2023-07-27T20:11:49Z

pubsub.go


 	ps.val.Start(ps)

 	go ps.processLoop(ctx)

-	(*PubSubNotif)(ps).Initialize()
+	// add current peers to notify system
+	notify.AddPeers(h.Network().Peers()...)


ah ok, it is still here.

This is racey; it needs to happen inside startMonitoring, after we have initialized the bus but before we have spawned the monitoring goroutine.

In the latest commit, the following test passes:

go test -v -run=TestSimpleDiscovery -count=1 .

As you suggest, we could move AddPeers to inside startMonitoring before the goroutine as p.AddPeers(p.host.Network().Peers()...) . But with this change, TestSimpleDiscovery hangs and times out. What do you think?

vyzo · 2023-07-27T20:12:38Z

notify.go

-func (p *PubSubNotif) Initialize() {
-	isTransient := func(pid peer.ID) bool {
-		for _, c := range p.host.Network().ConnsToPeer(pid) {
-			if !c.Stat().Transient {
-				return false
-			}


So the initialization code is still there (below), but I believe it is racey.

vyzo · 2023-07-27T20:13:36Z

notify.go

-func (p *PubSubNotif) Initialize() {
-	isTransient := func(pid peer.ID) bool {
-		for _, c := range p.host.Network().ConnsToPeer(pid) {
-			if !c.Stat().Transient {
-				return false
-			}


See comments below.

Also set eventbus.Name to "libp2p/pubsub/notify". Signed-off-by: Jeff Thompson <jeff@thefirst.org>

jefft0 · 2024-05-03T09:23:57Z

Hello @vyzo . You suggested moving AddPeers to inside startMonitoring before the goroutine as p.AddPeers(p.host.Network().Peers()...) . But with this change, TestSimpleDiscovery hangs and times out. What do you suggest to do to resolve this? (Right now we have to use a fork of this repo with the fix.)

vyzo · 2024-05-03T09:29:27Z

Uhm, fix the deadlock?

Maybe add an extra goroutine for startup.

fix: use event bus for monitoring peers connections and protocols update

15e9deb

Signed-off-by: gfanton <8671905+gfanton@users.noreply.github.com>

vyzo reviewed Jul 6, 2023

View reviewed changes

fix: split monitoring from add current peers

7aeb7da

Signed-off-by: Jeff Thompson <jeff@thefirst.org>

jefft0 force-pushed the fix/notify-protocol-update branch from e4c5794 to 7aeb7da Compare July 7, 2023 10:36

vyzo changed the title ~~fix: use event bus for monitoring peer connections and protocol updates~~ use event bus for monitoring peer connections and protocol updates Jul 7, 2023

vyzo requested changes Jul 27, 2023

View reviewed changes

vyzo reviewed Jul 27, 2023

View reviewed changes

On NotConnected event, call RemovePeers.

c710236

Also set eventbus.Name to "libp2p/pubsub/notify". Signed-off-by: Jeff Thompson <jeff@thefirst.org>

jefft0 requested a review from vyzo May 3, 2024 09:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

use event bus for monitoring peer connections and protocol updates #536

use event bus for monitoring peer connections and protocol updates #536

jefft0 commented Jul 6, 2023 •

edited

jefft0 commented Jul 6, 2023

vyzo left a comment •

edited

vyzo Jul 6, 2023

jefft0 Jul 7, 2023

jefft0 Jul 20, 2023

vyzo left a comment

vyzo Jul 27, 2023

vyzo Jul 27, 2023

vyzo Jul 27, 2023

vyzo Jul 27, 2023

jefft0 Jul 31, 2023

vyzo Jul 27, 2023

vyzo Jul 27, 2023

jefft0 Jul 31, 2023

vyzo Jul 27, 2023

vyzo Jul 27, 2023

jefft0 commented May 3, 2024

vyzo commented May 3, 2024

		if evt.Connectedness == network.Connected {
		go p.AddPeers(evt.Peer)

use event bus for monitoring peer connections and protocol updates #536

Are you sure you want to change the base?

use event bus for monitoring peer connections and protocol updates #536

Conversation

jefft0 commented Jul 6, 2023 • edited

jefft0 commented Jul 6, 2023

vyzo left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vyzo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jefft0 commented May 3, 2024

vyzo commented May 3, 2024

jefft0 commented Jul 6, 2023 •

edited

vyzo left a comment •

edited