Merge pull request #5200 from ipfs/fix/5197

fix dht commands when pubsub routing is enabled

core/commands/dht.go

@@ -13,7 +13,6 @@ import (
 	dag "github.com/ipfs/go-ipfs/merkledag"
 	path "github.com/ipfs/go-ipfs/path"
 
-	ipdht "gx/ipfs/QmNg6M98bwS97SL9ArvrRxKujFps3eV6XvmKgduiYga8Bn/go-libp2p-kad-dht"
 	routing "gx/ipfs/QmPpdpS9fknTBM3qHDcpayU6nYPZQeVjia2fbNrD8YWDe6/go-libp2p-routing"
 	notif "gx/ipfs/QmPpdpS9fknTBM3qHDcpayU6nYPZQeVjia2fbNrD8YWDe6/go-libp2p-routing/notifications"
 	b58 "gx/ipfs/QmWFAMPqsEyUX7gDUsRVmMWz59FxSpJ1b2v6bJ1yYzo7jY/go-base58-fast/base58"
@@ -26,6 +25,9 @@ import (
 
 var ErrNotDHT = errors.New("routing service is not a DHT")
 
+// TODO: Factor into `ipfs dht` and `ipfs routing`.
+// Everything *except `query` goes into `ipfs routing`.
+
 var DhtCmd = &cmds.Command{
 	Helptext: cmdkit.HelpText{
 		Tagline:          "Issue commands directly through the DHT.",
@@ -61,8 +63,7 @@ var queryDhtCmd = &cmds.Command{
 			return
 		}
 
-		dht, ok := n.Routing.(*ipdht.IpfsDHT)
+		if n.DHT == nil {
-		if !ok {
 			res.SetError(ErrNotDHT, cmdkit.ErrNormal)
 			return
 		}
@@ -76,7 +77,7 @@ var queryDhtCmd = &cmds.Command{
 			return
 		}
 
-		closestPeers, err := dht.GetClosestPeers(ctx, string(id))
+		closestPeers, err := n.DHT.GetClosestPeers(ctx, string(id))
 		if err != nil {
 			res.SetError(err, cmdkit.ErrNormal)
 			return
@@ -140,7 +141,7 @@ var queryDhtCmd = &cmds.Command{
 
 var findProvidersDhtCmd = &cmds.Command{
 	Helptext: cmdkit.HelpText{
-		Tagline:          "Find peers in the DHT that can provide a specific value, given a key.",
+		Tagline:          "Find peers that can provide a specific value, given a key.",
 		ShortDescription: "Outputs a list of newline-delimited provider Peer IDs.",
 	},
 
@@ -158,9 +159,8 @@ var findProvidersDhtCmd = &cmds.Command{
 			return
 		}
 
-		dht, ok := n.Routing.(*ipdht.IpfsDHT)
+		if n.Routing == nil {
-		if !ok {
+			res.SetError(errNotOnline, cmdkit.ErrNormal)
-			res.SetError(ErrNotDHT, cmdkit.ErrNormal)
 			return
 		}
 
@@ -186,7 +186,7 @@ var findProvidersDhtCmd = &cmds.Command{
 		outChan := make(chan interface{})
 		res.SetOutput((<-chan interface{})(outChan))
 
-		pchan := dht.FindProvidersAsync(ctx, c, numProviders)
+		pchan := n.Routing.FindProvidersAsync(ctx, c, numProviders)
 		go func() {
 			defer close(outChan)
 			for e := range events {
@@ -406,7 +406,7 @@ func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv ipld.DAGSe
 
 var findPeerDhtCmd = &cmds.Command{
 	Helptext: cmdkit.HelpText{
-		Tagline:          "Query the DHT for all of the multiaddresses associated with a Peer ID.",
+		Tagline:          "Find the multiaddresses associated with a Peer ID.",
 		ShortDescription: "Outputs a list of newline-delimited multiaddresses.",
 	},
 
@@ -423,9 +423,8 @@ var findPeerDhtCmd = &cmds.Command{
 			return
 		}
 
-		dht, ok := n.Routing.(*ipdht.IpfsDHT)
+		if n.Routing == nil {
-		if !ok {
+			res.SetError(errNotOnline, cmdkit.ErrNormal)
-			res.SetError(ErrNotDHT, cmdkit.ErrNormal)
 			return
 		}
 
@@ -454,7 +453,7 @@ var findPeerDhtCmd = &cmds.Command{
 
 		go func() {
 			defer close(events)
-			pi, err := dht.FindPeer(ctx, pid)
+			pi, err := n.Routing.FindPeer(ctx, pid)
 			if err != nil {
 				notif.PublishQueryEvent(ctx, &notif.QueryEvent{
 					Type:  notif.QueryError,
@@ -504,14 +503,14 @@ var findPeerDhtCmd = &cmds.Command{
 
 var getValueDhtCmd = &cmds.Command{
 	Helptext: cmdkit.HelpText{
-		Tagline: "Given a key, query the DHT for its best value.",
+		Tagline: "Given a key, query the routing system for its best value.",
 		ShortDescription: `
 Outputs the best value for the given key.
 
-There may be several different values for a given key stored in the DHT; in
+There may be several different values for a given key stored in the routing
-this context 'best' means the record that is most desirable. There is no one
+system; in this context 'best' means the record that is most desirable. There is
-metric for 'best': it depends entirely on the key type. For IPNS, 'best' is
+no one metric for 'best': it depends entirely on the key type. For IPNS, 'best'
-the record that is both valid and has the highest sequence number (freshest).
+is the record that is both valid and has the highest sequence number (freshest).
 Different key types can specify other 'best' rules.
 `,
 	},
@@ -529,9 +528,8 @@ Different key types can specify other 'best' rules.
 			return
 		}
 
-		dht, ok := n.Routing.(*ipdht.IpfsDHT)
+		if n.Routing == nil {
-		if !ok {
+			res.SetError(errNotOnline, cmdkit.ErrNormal)
-			res.SetError(ErrNotDHT, cmdkit.ErrNormal)
 			return
 		}
 
@@ -559,7 +557,7 @@ Different key types can specify other 'best' rules.
 
 		go func() {
 			defer close(events)
-			val, err := dht.GetValue(ctx, dhtkey)
+			val, err := n.Routing.GetValue(ctx, dhtkey)
 			if err != nil {
 				notif.PublishQueryEvent(ctx, &notif.QueryEvent{
 					Type:  notif.QueryError,
@@ -610,10 +608,10 @@ Different key types can specify other 'best' rules.
 
 var putValueDhtCmd = &cmds.Command{
 	Helptext: cmdkit.HelpText{
-		Tagline: "Write a key/value pair to the DHT.",
+		Tagline: "Write a key/value pair to the routing system.",
 		ShortDescription: `
 Given a key of the form /foo/bar and a value of any form, this will write that
-value to the DHT with that key.
+value to the routing system with that key.
 
 Keys have two parts: a keytype (foo) and the key name (bar). IPNS uses the
 /ipns keytype, and expects the key name to be a Peer ID. IPNS entries are
@@ -621,7 +619,7 @@ specifically formatted (protocol buffer).
 
 You may only use keytypes that are supported in your ipfs binary: currently
 this is only /ipns. Unless you have a relatively deep understanding of the
-go-ipfs DHT internals, you likely want to be using 'ipfs name publish' instead
+go-ipfs routing internals, you likely want to be using 'ipfs name publish' instead
 of this.
 
 Value is arbitrary text. Standard input can be used to provide value.
@@ -644,9 +642,8 @@ NOTE: A value may not exceed 2048 bytes.
 			return
 		}
 
-		dht, ok := n.Routing.(*ipdht.IpfsDHT)
+		if n.Routing == nil {
-		if !ok {
+			res.SetError(errNotOnline, cmdkit.ErrNormal)
-			res.SetError(ErrNotDHT, cmdkit.ErrNormal)
 			return
 		}
 
@@ -677,7 +674,7 @@ NOTE: A value may not exceed 2048 bytes.
 
 		go func() {
 			defer close(events)
-			err := dht.PutValue(ctx, key, []byte(data))
+			err := n.Routing.PutValue(ctx, key, []byte(data))
 			if err != nil {
 				notif.PublishQueryEvent(ctx, &notif.QueryEvent{
 					Type:  notif.QueryError,

core/core.go

@@ -41,7 +41,7 @@ import (
 	dhtopts "gx/ipfs/QmNg6M98bwS97SL9ArvrRxKujFps3eV6XvmKgduiYga8Bn/go-libp2p-kad-dht/opts"
 	u "gx/ipfs/QmPdKqUcHGFdeSpvjVoaTRPPstGif9GBZb5Q56RVw9o69A/go-ipfs-util"
 	routing "gx/ipfs/QmPpdpS9fknTBM3qHDcpayU6nYPZQeVjia2fbNrD8YWDe6/go-libp2p-routing"
-	psrouter "gx/ipfs/QmPxCZ99jTHMxD93qQV4pN3WJbBHrBLCQiumWDgfJjFQJy/go-libp2p-pubsub-router"
+	psrouter "gx/ipfs/QmR2hqcem4qjd4DkuyiwSFjfUiCP5eXHdPoM7o7dWKwct9/go-libp2p-pubsub-router"
 	pnet "gx/ipfs/QmRGvSwDpN4eunxgDNfmQhayZ6Z9F5a2v31V2D7y77osLg/go-libp2p-pnet"
 	goprocess "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess"
 	mamask "gx/ipfs/QmSMZwvs3n4GBikZ7hKzT17c3bk65FmyZo2JqtJ16swqCv/multiaddr-filter"
@@ -65,7 +65,7 @@ import (
 	ping "gx/ipfs/QmZ86eLPtXkQ1Dfa992Q8NpXArUoWWh3y728JDcWvzRrvC/go-libp2p/p2p/protocol/ping"
 	mplex "gx/ipfs/QmZHiqdRuNXujvSPNu1ZWxxzV6a2WhoZpfYkesdgyaKF9f/go-smux-multiplex"
 	pstore "gx/ipfs/QmZR2XWVVBCtbgBWnQhWk2xcQfaR3W8faQPriAiaaj7rsr/go-libp2p-peerstore"
-	rhelpers "gx/ipfs/QmZw5m4ioaoNmATBtP3o7qC1UERubJgz84RzccT3UEHZKr/go-libp2p-routing-helpers"
+	rhelpers "gx/ipfs/Qmafsgr3GSDKyGHW8SU9dbe6Vtv4rEgcgJ3WRnS72qtAzv/go-libp2p-routing-helpers"
 	cid "gx/ipfs/QmapdYm1b22Frv3k17fqrBYTFRxwiaVJkB299Mfn33edeB/go-cid"
 	p2phost "gx/ipfs/Qmb8T6YBBsjYsVGfrihQLfCJveczZnneSBqBKkYEBWDjge/go-libp2p-host"
 	logging "gx/ipfs/QmcVVHfdyv15GVPk7NrxdWjh2hLVccXnoD8j2tyQShiXJb/go-log"
@@ -137,6 +137,7 @@ type IpfsNode struct {
 
 	Floodsub *floodsub.PubSub
 	PSRouter *psrouter.PubsubValueStore
+	DHT      *dht.IpfsDHT
 	P2P      *p2p.P2P
 
 	proc goprocess.Process
@@ -465,6 +466,23 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost
 	}
 	n.Routing = r
 
+	// TODO: I'm not a fan of type assertions like this but the
+	// `RoutingOption` system doesn't currently provide access to the
+	// IpfsNode.
+	//
+	// Ideally, we'd do something like:
+	//
+	// 1. Add some fancy method to introspect into tiered routers to extract
+	//    things like the pubsub router or the DHT (complicated, messy,
+	//    probably not worth it).
+	// 2. Pass the IpfsNode into the RoutingOption (would also remove the
+	//    PSRouter case below.
+	// 3. Introduce some kind of service manager? (my personal favorite but
+	//    that requires a fair amount of work).
+	if dht, ok := r.(*dht.IpfsDHT); ok {
+		n.DHT = dht
+	}
+
 	if ipnsps {
 		n.PSRouter = psrouter.NewPubsubValueStore(
 			ctx,
@@ -601,8 +619,8 @@ func (n *IpfsNode) teardown() error {
 		closers = append(closers, mount.Closer(n.Mounts.Ipns))
 	}
 
-	if dht, ok := n.Routing.(*dht.IpfsDHT); ok {
+	if n.DHT != nil {
-		closers = append(closers, dht.Process())
+		closers = append(closers, n.DHT.Process())
 	}
 
 	if n.Blocks != nil {

package.json

@@ -524,15 +524,15 @@
     },
     {
       "author": "stebalien",
-      "hash": "QmPxCZ99jTHMxD93qQV4pN3WJbBHrBLCQiumWDgfJjFQJy",
+      "hash": "QmR2hqcem4qjd4DkuyiwSFjfUiCP5eXHdPoM7o7dWKwct9",
       "name": "go-libp2p-pubsub-router",
-      "version": "0.3.5"
+      "version": "0.3.6"
     },
     {
       "author": "Stebalien",
-      "hash": "QmZw5m4ioaoNmATBtP3o7qC1UERubJgz84RzccT3UEHZKr",
+      "hash": "Qmafsgr3GSDKyGHW8SU9dbe6Vtv4rEgcgJ3WRnS72qtAzv",
       "name": "go-libp2p-routing-helpers",
-      "version": "0.2.4"
+      "version": "0.2.5"
     },
     {
       "author": "fsnotify",

test/sharness/lib/iptb-lib.sh

@@ -35,9 +35,9 @@ startup_cluster() {
   bound=$(expr "$num_nodes" - 1)
 
   if test -n "$other_args"; then
-    test_expect_success "start up nodes with additional args" '
+    test_expect_success "start up nodes with additional args" "
-      iptb start --args $other_args
+      iptb start --args \"${other_args[@]}\"
-    '
+    "
   else
     test_expect_success "start up nodes" '
       iptb start

test/sharness/t0170-dht.sh

@@ -7,13 +7,15 @@ test_description="Test dht command"
 TEST_DHT_VALUE="foobar"
 TEST_DHT_PATH="/pk/QmbWTwYGcmdyK9CYfNBcfs9nhZs17a6FQ4Y8oea278xx41"
 
-# start iptb + wait for peering
+test_dht() {
   NUM_NODES=5
+
   test_expect_success 'init iptb' '
+    rm -rf .iptb/ &&
     iptb init -n $NUM_NODES --bootstrap=none --port=0
   '
 
-startup_cluster $NUM_NODES
+  startup_cluster $NUM_NODES "$@"
 
   test_expect_success 'peer ids' '
     PEERID_0=$(iptb get id 0) &&
@@ -89,5 +91,9 @@ test_expect_success 'query' '
   test_expect_success 'stop iptb' '
     iptb stop
   '
+}
+
+test_dht
+test_dht --enable-pubsub-experiment --enable-namesys-pubsub
 
 test_done