fix: remove pubsub discovery hack

Pubsub handles this internally now.

Author: Stebalien

core/commands/pubsub.go

@@ -74,7 +74,7 @@ This command outputs data in the following encodings:
 		cmds.StringArg("topic", true, false, "String name of topic to subscribe to."),
 	},
 	Options: []cmds.Option{
-		cmds.BoolOption(pubsubDiscoverOptionName, "try to discover other peers subscribed to the same topic"),
+		cmds.BoolOption(pubsubDiscoverOptionName, "Deprecated option to instruct pubsub to discovery peers for the topic. Discovery is now built into pubsub."),
 	},
 	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
 		api, err := cmdenv.GetApi(env, req)
@@ -83,9 +83,7 @@ This command outputs data in the following encodings:
 		}
 
 		topic := req.Arguments[0]
-		discover, _ := req.Options[pubsubDiscoverOptionName].(bool)
-
-		sub, err := api.PubSub().Subscribe(req.Context, topic, options.PubSub.Discover(discover))
+		sub, err := api.PubSub().Subscribe(req.Context, topic)
 		if err != nil {
 			return err
 		}

core/coreapi/pubsub.go

@@ -3,7 +3,6 @@ package coreapi
 import (
 	"context"
 	"errors"
-	"strings"
 	"sync"
 	"time"
 
@@ -19,7 +18,6 @@ import (
 type PubSubAPI CoreAPI
 
 type pubSubSubscription struct {
-	cancel       context.CancelFunc
 	subscription *pubsub.Subscription
 }
 
@@ -61,12 +59,11 @@ func (api *PubSubAPI) Publish(ctx context.Context, topic string, data []byte) er
 }
 
 func (api *PubSubAPI) Subscribe(ctx context.Context, topic string, opts ...caopts.PubSubSubscribeOption) (coreiface.PubSubSubscription, error) {
-	options, err := caopts.PubSubSubscribeOptions(opts...)
-	if err != nil {
-		return nil, err
-	}
-
-	r, err := api.checkNode()
+	// Parse the options to avoid introducing silent failures for invalid
+	// options. However, we don't currently have any use for them. The only
+	// subscription option, discovery, is now a no-op as it's handled by
+	// pubsub itself.
+	_, err := caopts.PubSubSubscribeOptions(opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -77,21 +74,7 @@ func (api *PubSubAPI) Subscribe(ctx context.Context, topic string, opts ...caopt
 		return nil, err
 	}
 
-	pubctx, cancel := context.WithCancel(api.nctx)
-
-	if options.Discover {
-		go func() {
-			blk, err := api.core().Block().Put(pubctx, strings.NewReader("floodsub:"+topic))
-			if err != nil {
-				log.Error("pubsub discovery: ", err)
-				return
-			}
-
-			connectToPubSubPeers(pubctx, r, api.peerHost, blk.Path().Cid())
-		}()
-	}
-
-	return &pubSubSubscription{cancel, sub}, nil
+	return &pubSubSubscription{sub}, nil
 }
 
 func connectToPubSubPeers(ctx context.Context, r routing.Routing, ph p2phost.Host, cid cid.Cid) {
@@ -132,7 +115,6 @@ func (api *PubSubAPI) checkNode() (routing.Routing, error) {
 }
 
 func (sub *pubSubSubscription) Close() error {
-	sub.cancel()
 	sub.subscription.Cancel()
 	return nil
 }