bug: Messages won't get through with multiple pubsub topics

[romanzac]

When I run white box test with multiple (2) pubsub topics, I am not able to receive messages which I've sent from relay node.

Steps to reproduce the behavior:

1. Checkout branch "chore(filterV2)-test-updates"
2. Run the test
   go test -run ^TestFilterSuite$ -testify.m TestMultiPubSubMultiContentTopic
3. No error is visible. s.relayNode.PublishToTopic() call won't return any error.

Expected behavior
Messages get received by light node.

Log
go test -run ^TestFilterSuite$ -testify.m TestMultiPubSubMultiContentTopic
2023-10-13T09:37:59.880+0800 INFO gowaku.filterv2-lightnode filter/client.go:84 filter-push protocol started
2023-10-13T09:37:59.997+0800 INFO gowaku.relay relay/waku_relay.go:254 Relay protocol started
2023-10-13T09:37:59.997+0800 INFO gowaku.relay relay/waku_relay.go:344 subscribing to topic {"topic": "/waku/2/go/filter/test"}
2023-10-13T09:37:59.997+0800 INFO gowaku.filterv2-fullnode filter/server.go:82 filter-subscriber protocol started
2023-10-13T09:37:59.997+0800 INFO gowaku filter/filter_test.go:823 Executing {"testName": "TestMultiPubSubMultiContentTopic"}
2023-10-13T09:38:00.024+0800 INFO gowaku.filterv2-fullnode filter/server.go:118 received request {"peer": "QmQvGz6DvdTYvgmQ6wDUaCZiddb226BbPrAdAiWe7Nax2x", "requestID": "36a279ebd5dd77e5a11527dc9d7726c689b47d3b94340ff48de351650e7fb721", "requestType": "SUBSCRIBE"}
2023-10-13T09:38:01.025+0800 INFO gowaku filter/filter_test.go:803 Subscribing {"PubSubTopic": "/waku/2/go/filter/test00"}
2023-10-13T09:38:01.029+0800 INFO gowaku.filterv2-fullnode filter/server.go:118 received request {"peer": "QmQvGz6DvdTYvgmQ6wDUaCZiddb226BbPrAdAiWe7Nax2x", "requestID": "8bd9799940406f9d69e8b9416de148a70db74cd4040d52f62e1694fcaa44ddfc", "requestType": "SUBSCRIBE"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:803 Subscribing {"PubSubTopic": "/waku/2/go/filter/test01"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:808 Light Node subscription {"PubSubTopic": "/waku/2/go/filter/test00"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:809 Light Node subscription {"ContentTopic": "Topic00"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:808 Light Node subscription {"PubSubTopic": "/waku/2/go/filter/test01"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:809 Light Node subscription {"ContentTopic": "Topic01"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:153 Expected messages {"count": "2"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:154 Existing subscriptions {"count": "2"}
2023-10-13T09:38:02.030+0800 INFO gowaku filter/filter_test.go:159 Looking at {"pubSubTopic": "/waku/2/go/filter/test00"}
2023-10-13T09:38:06.031+0800 INFO gowaku filter/filter_test.go:159 Looking at {"pubSubTopic": "/waku/2/go/filter/test01"}
2023-10-13T09:38:10.035+0800 INFO gowaku filter/filter_test.go:827 Finished executing {"testName": "TestMultiPubSubMultiContentTopic"}
2023-10-13T09:38:10.041+0800 INFO pubsub go-libp2p-pubsub@v0.9.3/pubsub.go:671 pubsub processloop shutting down
2023-10-13T09:38:10.042+0800 ERROR gowaku.filterv2-lightnode filter/client.go:196 receiving FilterSubscribeResponse {"error": "protocols not supported: [/vac/waku/filter-subscribe/2.0.0-beta1]"}
2023-10-13T09:38:10.042+0800 ERROR gowaku.filterv2-lightnode filter/client.go:563 could not unsubscribe from peer {"peerID": "QmZF3v5SZPVPZs7L4V2WrqapbSonbxwxH2RmwL9wW7cbKK", "error": "protocols not supported: [/vac/waku/filter-subscribe/2.0.0-beta1]"}
2023-10-13T09:38:10.042+0800 WARN gowaku.filterv2-lightnode filter/client.go:99 unsubscribing from full nodes {"error": "protocols not supported: [/vac/waku/filter-subscribe/2.0.0-beta1]", "peerID": "QmZF3v5SZPVPZs7L4V2WrqapbSonbxwxH2RmwL9wW7cbKK"}
--- FAIL: TestFilterSuite (10.40s)
--- FAIL: TestFilterSuite/TestMultiPubSubMultiContentTopic (10.40s)
filter_test.go:188:
Error Trace: /Users/roman/sources/waku-org/go-waku/waku/v2/protocol/filter/filter_test.go:188
/Users/roman/sources/waku-org/go-waku/waku/v2/protocol/filter/filter_test.go:813
Error: Should be true
Test: TestFilterSuite/TestMultiPubSubMultiContentTopic
FAIL
exit status 1
FAIL github.com/waku-org/go-waku/waku/v2/protocol/filter 11.296s

[chaitanyaprem]

Went through the test code and i had noticed that minPeersToPublish is set to 0 when initializing WakuRelay here

go-waku/waku/v2/protocol/filter/filter_test.go

Line 60 in 2ef7e73

relay := relay.NewWakuRelay(broadcaster, 0, timesource.NewDefaultClock(), prometheus.DefaultRegisterer, s.log)

This means the call for publishToTopic would just publish message even if there are no peers connected for that topic.
We need to change minPeersToPublish to 1 and then wait for atleast 1 peer to be connected to the relay node for each pubSubTopic.
Post which if we try to publish, you should start seeing messages being delivered properly.

@richard-ramos , wondering if should not allow setting this to 0 in order to avoid this kind of a situation. Is there any use-case you can think of where such an initialization of relay would be required i.e publishing should be allowed even with 0? If not, we can probably disallow setting this to 0.

Ignore the above, that is not relevant.
Issue could be that broadcaster is registered for only the pubSubTopic that relay was initialized with which might be causing the relay to not broadcast messages for other pubSubTopics in the broadcaster which are subscribed later.
Since FilterFullnode is initialized out of the broadcaster, maybe registering broadcaster for all pubSubTopics solves the issue.

[romanzac]

I've tried to call broadcaster.RegisterForAll(), still with the same result. Is there any other way how to start full node with more pubSubTopics ?

node2Filter := NewWakuFilterFullNode(timesource.NewDefaultClock(), prometheus.DefaultRegisterer, s.log)
node2Filter.SetHost(host)
//sub := broadcaster.Register(topic)
sub := broadcaster.RegisterForAll()
err := node2Filter.Start(s.ctx, sub)

I would image light node being limited to just one pubSub topic. Not expecting this limitation from full node. Do you have similar view ?

R.

[chaitanyaprem]

I've tried to call broadcaster.RegisterForAll(), still with the same result. Is there any other way how to start full node with more pubSubTopics ?

Went through the code again and realized that the relay for the fullNode is not subscribed to all the topics which is why messages are not being received by fullNode which inturn are not sent to lightNode.
I have added the following lines marked with + in your message loop where you subscribe for lightNode and then it started working. Pushing the changes to the branch as well.

        for _, m := range messages {
                s.subDetails = append(s.subDetails, s.subscribe(m.pubSubTopic, m.contentTopic, s.fullNodeHost.ID())...)
                s.log.Info("Subscribing ", zap.String("PubSubTopic", m.pubSubTopic))

+               _, err := s.relayNode.SubscribeToTopic(context.Background(), m.pubSubTopic)
+               s.Require().NoError(err)

Yes, we can pass the pubSubTopics that we want the fullNode to be part of by subscribing the relay to these topics.

node2Filter := NewWakuFilterFullNode(timesource.NewDefaultClock(), prometheus.DefaultRegisterer, s.log)
node2Filter.SetHost(host)
//sub := broadcaster.Register(topic)
sub := broadcaster.RegisterForAll()
err := node2Filter.Start(s.ctx, sub)

I would image light node being limited to just one pubSub topic. Not expecting this limitation from full node. Do you have similar view ?

R.

Not necessarily restricted to one pubSubTopic. Especially if we consider autosharding and let's say an app is using different contentTopics which fall under different shard(which means different pubSubTopics) then the lightNode would be subscribing to multiple pubSubTopics.
Do note that lightNode(FilterLightNode) is anyway working with just request-response Filter protocol so 1 or many pubSubTopics wouldn't matter.
Hope i made it clear.

[romanzac]

I would like to subscribe to new pubsubTopics from light node, not from relay node. TestMultiPubSubMultiContentTopic targets scenario, when I run light node only and have no programatic access to relay nor full node. I use white box approach for now, so it might be not obvious what do I expect from the test result. Sorry for that.

Additionally RegisterForAll() looks contrarian to the issue #810 we have just fixed to prevent subscribe with empty ("") topics. Could I be already subscribed to pubsub topic even I have never sent request for this particular pubsub before ? If this is the case, I see a place for an attacker to exploit. It might also be RegisterForAll() has another purpose, I am not aware of yet ?

// RegisterForAll returns a subscription for all topics
func (b *broadcaster) RegisterForAll(chLen ...int) Subscription {
return b.chStore.getNewCh("", getChLen(chLen))
}

[chaitanyaprem]

I would like to subscribe to new pubsubTopics from light node, not from relay node. TestMultiPubSubMultiContentTopic targets scenario, when I run light node only and have no programatic access to relay nor full node. I use white box approach for now, so it might be not obvious what do I expect from the test result. Sorry for that.

Understood that, but the light-node is subscribing to a filterFullNode which is also acting as a relay node. Which means the relay node also should be subscribed to the pubsubTopics(that are of interest to lightNode) so that it can receive the messages flowing in those pubsubTopics and also forward them to the lightNode which has expressed interest.

No problem, do note that you are writing a unit test of the protocol and hence require simulating required fullNode etc around it for the test to be complete.

Additionally RegisterForAll() looks contrarian to the issue #810 we have just fixed to prevent subscribe with empty ("") topics.

Issue #810 was for filter client where a client cannot subscribe with a contentFilter having empty content-Topics because filter protocol doesn't support a filter-subscription without specifying a content-topic. This issue was a corner case where an empty value was specified for contentTopic rather than not setting it.

If you look carefully the RegisterForAll() is being done not at filterClient side of the code whereas it is done at filterFullNode indicating that we want to register for all pubsubTopics, whereas issue #810 is for filterClient side of the API indicating empty contentTopic is not allowed.

Could I be already subscribed to pubsub topic even I have never sent request for this particular pubsub before ?

Are you talking about the filterClient side of things here or fullNode? As i had explained above, the test is simulating a filterFullNode and a filterClientNode in order to test the filterProtocol. In filterClient node, you cannot be subscribed to a pubsubTopic without sending a filter-subscribe request to a fullNode.

If this is the case, I see a place for an attacker to exploit. It might also be RegisterForAll() has another purpose, I am not aware of yet ?
// RegisterForAll returns a subscription for all topics func (b *broadcaster) RegisterForAll(chLen ...int) Subscription { return b.chStore.getNewCh("", getChLen(chLen)) }

RegisterForAll and broadcaster is required when you want your node (be it relay, filterFullNode or a filterClient/lightNode) to also act as a storeNode. This broadcaster is used to receive messages and then through the subscription pass on to store protocol so that they get stored in the DB. The broadcaster and Register logic is quite confusing and even it took time for me to figure it out. Hope i was able to help clarify it for you.

[romanzac]

We can close this issue with conclusions we have made during our conf call:
#804 - Test design should change as light-node could subscribe to more than one pubsubTopic only if its peer full node is already subscribed to these topics.
#810 - There is no visible risk of exploiting "broadcaster.RegisterForAll()" as subscription to the pubsubTopic is still required.

[chaitanyaprem]

#810 - There is no visible risk of exploiting "broadcaster.RegisterForAll()" as subscription to the pubsubTopic is still required.

Just to add more context as i explained during the call, broadcaster is an internal implementation detail of go-waku and RegisterForAll is just registering with relay to stream messages belonging to all locally subscribed pubsubTopics through it. Hence it cannot be exploited as the serviceNode chose to support the subscribed pubsubTopics in the Waku Network.