Make pubsub optional

[dgtony]

Running nodes for some time we've detected that pubsub mechanism combined with threads may produce a lot of unneccesary traffic.
Assuming that every thread has corresponding topic propagated through the pubsub, long-running high-degree nodes accumulate a lot of topics during its lifetime. When some other node establishes connection, it will respond with a huge hello-packet, containing all the topics seen so far, mostly irrelevant to the caller.
Taking into account that pubsub is in fact a secondary channel of event propagation, we can alleviate the problem by making it optional. Pubsub stays enabled by default, so the change won't affect any existing setup, and there is a new NetOption allowing to disable pubsub subsystem in threads completely.

[sanderpick]

Thanks for jumping in here! I left a couple comments around semantics, otherwise LGTM.

[sanderpick]

How about WithNetPubsub, and when setting up NetConfig, set PubSub to true by default?

[sanderpick]

Opps, I forgot NetConfig is public. In any case, given your findings, I'm fine with letting pubsub be off by default. So, same comment as before around semantics, but no need to have PubSub true by default.

[sanderpick]

(We can do a minor version bump)

[dgtony]

Sure, now it looks more consistent.

[sanderpick]

To go along with the comment below, this would just be PubSub.

[dgtony]

Fixed

[sanderpick]

Great! For some reason I assumed this wouldn't be desired at the DB level.

[dgtony]

It actually slipped into the PR, but anyway we have db failures without it.

[sanderpick]

One more nit: Can you make this more like the debug option that makes using an upstream bool a little nicer, as in: WithNetPubSub(enabled bool)

[dgtony]

Fixed

[sanderpick]

Looks like test failures from pubsub not being default. Can you git apply this patch?

diff --git a/api/client/client_test.go b/api/client/client_test.go
index 96545a3..bf1b39e 100644
--- a/api/client/client_test.go
+++ b/api/client/client_test.go
@@ -746,7 +746,11 @@ func makeServer(t *testing.T) (ma.Multiaddr, func()) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	n, err := common.DefaultNetwork(dir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		dir,
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetPubSub(true),
+		common.WithNetDebug(true))
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/examples/chat/main.go b/examples/chat/main.go
index d727463..5a2bcb4 100644
--- a/examples/chat/main.go
+++ b/examples/chat/main.go
@@ -96,6 +96,7 @@ func main() {
 	net, err = common.DefaultNetwork(
 		*repo,
 		common.WithNetHostAddr(hostAddr),
+		common.WithNetPubSub(true),
 		common.WithNetDebug(*debug))
 	if err != nil {
 		log.Fatal(err)
diff --git a/net/api/client/client_test.go b/net/api/client/client_test.go
index 6ae6b31..3fc0b83 100644
--- a/net/api/client/client_test.go
+++ b/net/api/client/client_test.go
@@ -415,6 +415,7 @@ func makeServer(t *testing.T) (ma.Multiaddr, ma.Multiaddr, func()) {
 	n, err := common.DefaultNetwork(
 		dir,
 		common.WithNetHostAddr(hostAddr),
+		common.WithNetPubSub(true),
 		common.WithNetDebug(true))
 	if err != nil {
 		t.Fatal(err)
diff --git a/threadsd/main.go b/threadsd/main.go
index f0a9348..186dfc2 100644
--- a/threadsd/main.go
+++ b/threadsd/main.go
@@ -36,6 +36,7 @@ func main() {
 	connHighWater := fs.Int("connHighWater", 400, "High watermark of libp2p connections that'll be maintained")
 	connGracePeriod := fs.Duration("connGracePeriod", time.Second*20, "Duration a newly opened connection is given before it becomes subject to pruning")
 	keepAliveInterval := fs.Duration("keepAliveInterval", time.Second*5, "Enables websocket keepalive pinging with the specified timeout interval. Configured interval must be >= 1s.")
+	enableNetPubsub := fs.Bool("enableNetPubsub", false, "Enable thread networking over libp2p pubsub")
 	debug := fs.Bool("debug", false, "Enable debug logging")
 	if err := fs.Parse(os.Args[1:]); err != nil {
 		log.Fatal(err)
@@ -65,12 +66,18 @@ func main() {
 	log.Debugf("hostAddr: %v", *hostAddrStr)
 	log.Debugf("apiAddr: %v", *apiAddrStr)
 	log.Debugf("apiProxyAddr: %v", *apiProxyAddrStr)
+	log.Debugf("connLowWater: %v", *connLowWater)
+	log.Debugf("connHighWater: %v", *connHighWater)
+	log.Debugf("connGracePeriod: %v", *connGracePeriod)
+	log.Debugf("keepAliveInterval: %v", *keepAliveInterval)
+	log.Debugf("enableNetPubsub: %v", *enableNetPubsub)
 	log.Debugf("debug: %v", *debug)

 	n, err := common.DefaultNetwork(
 		*repo,
 		common.WithNetHostAddr(hostAddr),
 		common.WithConnectionManager(connmgr.NewConnManager(*connLowWater, *connHighWater, *connGracePeriod)),
+		common.WithNetPubSub(*enableNetPubsub),
 		common.WithNetDebug(*debug))
 	if err != nil {
 		log.Fatal(err)

[dgtony]

Maybe it's better to make a major version bump as it changes default behavior?

[sanderpick]

Yeah, good idea. Do you need a release right away?

[dgtony]

No, it's not urgent.