Add optional libp2p discovery (#3323)

use -libp2p-discovery as daemon arg to enable. There is a -libp2p-port argument but it is IGNORED at the moment in favor of the hardcoded port 28675 (arbitrary). When we remove the Haskell Kademlia, we'll take over the existing port infrastructure.

There's a -disable-old-discovery that turns off old discovery as well. Risky!

* cli flags

* Gossip_net: keep the whole Config.t

* add libp2p_membership field

* missing opam file

* start threading it all through

* big ole nasty commit that does a bunch

* libp2p keypair + initial peers

* inject new libp2p peers back into haskell membership

Author: Corey Richardson

src/codanet.go

@@ -4,47 +4,54 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"log"
+	"os"
+	"path"
+	"strconv"
+	"time"
+
 	dsb "github.com/ipfs/go-ds-badger"
+	logging "github.com/ipfs/go-log"
 	p2p "github.com/libp2p/go-libp2p"
 	crypto "github.com/libp2p/go-libp2p-core/crypto"
 	host "github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-core/peerstore"
 	routing "github.com/libp2p/go-libp2p-core/routing"
 	discovery "github.com/libp2p/go-libp2p-discovery"
 	kad "github.com/libp2p/go-libp2p-kad-dht"
 	kadopts "github.com/libp2p/go-libp2p-kad-dht/opts"
-	"github.com/libp2p/go-libp2p-peerstore"
 	"github.com/libp2p/go-libp2p-peerstore/pstoreds"
 	pnet "github.com/libp2p/go-libp2p-pnet"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
-	"github.com/libp2p/go-libp2p-record"
+	record "github.com/libp2p/go-libp2p-record"
 	secio "github.com/libp2p/go-libp2p-secio"
 	p2pconfig "github.com/libp2p/go-libp2p/config"
 	mdns "github.com/libp2p/go-libp2p/p2p/discovery"
 	tcp "github.com/libp2p/go-tcp-transport"
 	ws "github.com/libp2p/go-ws-transport"
 	"github.com/multiformats/go-multiaddr"
 	"golang.org/x/crypto/blake2b"
-	"log"
-	"os"
-	"path"
-	"strconv"
-	"time"
 )
 
 // Helper contains all the daemon state
 type Helper struct {
-	Host   host.Host
-	Mdns   mdns.Service
-	Dht    *kad.IpfsDHT
-	Ctx    context.Context
-	Pubsub *pubsub.PubSub
+	Host            host.Host
+	Mdns            mdns.Service
+	Dht             *kad.IpfsDHT
+	Ctx             context.Context
+	Pubsub          *pubsub.PubSub
+	Logger          logging.EventLogger
+	DiscoveredPeers chan peer.AddrInfo
+	Rendezvous      string
+	Discovery       *discovery.RoutingDiscovery
 }
 
 type mdnsListener struct {
-	FoundPeer chan peerstore.PeerInfo
+	FoundPeer chan peer.AddrInfo
 }
 
-func (l *mdnsListener) HandlePeerFound(info peerstore.PeerInfo) {
+func (l *mdnsListener) HandlePeerFound(info peer.AddrInfo) {
 	l.FoundPeer <- info
 }
 
@@ -64,6 +71,7 @@ func (cv customValidator) Select(key string, values [][]byte) (int, error) {
 
 // MakeHelper does all the initialization to run one host
 func MakeHelper(ctx context.Context, listenOn []multiaddr.Multiaddr, statedir string, pk crypto.PrivKey, networkID string) (*Helper, error) {
+	logger := logging.Logger("codanet.Helper")
 	dso := dsb.DefaultOptions
 
 	bp := path.Join(statedir, strconv.Itoa(os.Getpid()))
@@ -85,14 +93,17 @@ func MakeHelper(ctx context.Context, listenOn []multiaddr.Multiaddr, statedir st
 		return nil, err
 	}
 
-	pnetKey := blake2b.Sum256([]byte("/coda/0.0.1"))
+	rendezvousString := fmt.Sprintf("/coda/0.0.1/%s", networkID)
+
+	pnetKey := blake2b.Sum256([]byte(rendezvousString))
 	prot, err := pnet.NewV1ProtectorFromBytes(&pnetKey)
 	if err != nil {
 		return nil, err
 	}
 
 	rv := customValidator{Base: record.NamespacedValidator{"pk": record.PublicKeyValidator{}}}
 
+	// gross hack to exfiltrate a channel from the side effect of option evaluation
 	kadch := make(chan *kad.IpfsDHT)
 
 	// Make sure this doesn't get too out of sync with the defaults,
@@ -118,13 +129,11 @@ func MakeHelper(ctx context.Context, listenOn []multiaddr.Multiaddr, statedir st
 		return nil, err
 	}
 
-	rendezvousString := fmt.Sprintf("/coda/0.0.1/%s", networkID)
-
 	mdns, err := mdns.NewMdnsService(ctx, host, time.Minute, "_coda-discovery._udp.local")
 	if err != nil {
 		return nil, err
 	}
-	l := &mdnsListener{FoundPeer: make(chan peerstore.PeerInfo)}
+	l := &mdnsListener{FoundPeer: make(chan peer.AddrInfo)}
 	mdns.RegisterNotifee(l)
 
 	kad := <-kadch
@@ -135,44 +144,60 @@ func MakeHelper(ctx context.Context, listenOn []multiaddr.Multiaddr, statedir st
 	routingDiscovery := discovery.NewRoutingDiscovery(kad)
 
 	log.Println("Announcing ourselves for", rendezvousString)
-	discovery.Advertise(ctx, routingDiscovery, rendezvousString)
 
-	// try and find some peers for this chain
-	//dhtpeers, err := routingDiscovery.FindPeers(ctx, rendezvousString, discovery.Limit(16))
-	//if err != nil {
-	//	return nil, err
-	//}
+	discovered := make(chan peer.AddrInfo)
 
-	foundPeer := func(info peerstore.PeerInfo, source string) {
+	foundPeer := func(info peer.AddrInfo, source string) {
 		if info.ID != "" {
 			ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
 			defer cancel()
 			if err := host.Connect(ctx, info); err != nil {
-				log.Printf("Warn: couldn't connect to %s peer %v (different chain?): %v", source, info.Loggable(), err)
+				logger.Warning("couldn't connect to %s peer %v (maybe the network ID mismatched?): %v", source, info.Loggable(), err)
 			} else {
-				log.Printf("Found a %s peer: %s", source, info.Loggable())
-				host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL)
+				logger.Info("Found a %s peer: %s", source, info.Loggable())
+				host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.ConnectedAddrTTL)
+				discovered <- info
 			}
 		}
 	}
 
+	// report local discovery peers
 	go func() {
-		for {
-			info := <-l.FoundPeer
+		for info := range l.FoundPeer {
 			foundPeer(info, "local")
 		}
 	}()
 
+	// report dht peers
+	go func() {
+		for {
+			// default is to yield only 100 peers at a time. for now, always be
+			// looking... TODO: Is there a better way to use discovery? Should we only
+			// have to explicitly search once?
+			dhtpeers, err := routingDiscovery.FindPeers(ctx, rendezvousString)
+			if err != nil {
+				logger.Error("failed to find DHT peers: ", err)
+			}
+			for info := range dhtpeers {
+				foundPeer(info, "dht")
+			}
+		}
+	}()
+
 	pubsub, err := pubsub.NewFloodSub(ctx, host, pubsub.WithStrictSignatureVerification(true), pubsub.WithMessageSigning(true))
 	if err != nil {
 		return nil, err
 	}
 
 	return &Helper{
-		Host:   host,
-		Ctx:    ctx,
-		Mdns:   mdns,
-		Dht:    kad,
-		Pubsub: pubsub,
+		Host:            host,
+		Ctx:             ctx,
+		Mdns:            mdns,
+		Dht:             kad,
+		Pubsub:          pubsub,
+		Logger:          logger,
+		DiscoveredPeers: discovered,
+		Rendezvous:      rendezvousString,
+		Discovery:       routingDiscovery,
 	}, nil
 }

src/generate_methodidx/main.go

@@ -99,7 +99,7 @@ func main() {
 		Command:     "generate_methodidx",
 		PackageName: "main",
 		TypesAndValues: map[string][]string{
-			"methodIdx": []string{"configure", "listen", "publish", "subscribe", "unsubscribe",  "validationComplete", "generateKeypair", "openStream", "closeStream", "resetStream", "sendStreamMsg", "removeStreamHandler", "addStreamHandler", "listeningAddrs"},
+			"methodIdx": []string{"configure", "listen", "publish", "subscribe", "unsubscribe", "validationComplete", "generateKeypair", "openStream", "closeStream", "resetStream", "sendStreamMsg", "removeStreamHandler", "addStreamHandler", "listeningAddrs", "addPeer", "beginAdvertising"},
 		},
 	}
 

src/libp2p_helper/main.go

@@ -20,6 +20,7 @@ import (
 	net "github.com/libp2p/go-libp2p-core/network"
 	peer "github.com/libp2p/go-libp2p-core/peer"
 	protocol "github.com/libp2p/go-libp2p-core/protocol"
+	discovery "github.com/libp2p/go-libp2p-discovery"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	b58 "github.com/mr-tron/base58/base58"
 	"github.com/multiformats/go-multiaddr"
@@ -64,6 +65,8 @@ const (
 	removeStreamHandler
 	addStreamHandler
 	listeningAddrs
+	addPeer
+	beginAdvertising
 )
 
 type envelope struct {
@@ -134,6 +137,12 @@ type configureMsg struct {
 	ListenOn  []string `json:"ifaces"`
 }
 
+type discoveredPeerUpcall struct {
+	ID     string   `json:"peer_id"`
+	Addrs  []string `json:"multiaddrs"`
+	Upcall string   `json:"upcall"`
+}
+
 func (m *configureMsg) run(app *app) (interface{}, error) {
 	privkBytes, err := b58.Decode(m.Privk)
 	if err != nil {
@@ -152,6 +161,19 @@ func (m *configureMsg) run(app *app) (interface{}, error) {
 		maddrs[i] = res
 	}
 	helper, err := codanet.MakeHelper(app.Ctx, maddrs, m.Statedir, privk, m.NetworkID)
+	go func() {
+		for info := range helper.DiscoveredPeers {
+			addrStrings := make([]string, len(info.Addrs))
+			for i, a := range info.Addrs {
+				addrStrings[i] = a.String()
+			}
+			app.writeMsg(discoveredPeerUpcall{
+				ID:     peer.IDB58Encode(info.ID),
+				Addrs:  addrStrings,
+				Upcall: "discoveredPeer",
+			})
+		}
+	}()
 	if err != nil {
 		return nil, badHelper(err)
 	}
@@ -546,6 +568,52 @@ func (rs *removeStreamHandlerMsg) run(app *app) (interface{}, error) {
 	return "removeStreamHandler success", nil
 }
 
+type addPeerMsg struct {
+	Multiaddr string `json:"multiaddr"`
+}
+
+func (ap *addPeerMsg) run(app *app) (interface{}, error) {
+	if app.P2p == nil {
+		return nil, needsConfigure()
+	}
+	multiaddr, err := multiaddr.NewMultiaddr(ap.Multiaddr)
+	if err != nil {
+		// TODO: this isn't necessarily an RPC error. Perhaps the encoded multiaddr
+		// isn't supported by this version of libp2p.
+		// But more likely, it is an RPC error.
+		return nil, badRPC(err)
+	}
+	info, err := peer.AddrInfoFromP2pAddr(multiaddr)
+	if err != nil {
+		// TODO: this isn't necessarily an RPC error. Perhaps the contained peer ID
+		// isn't supported by this version of libp2p.
+		// But more likely, it is an RPC error.
+		return nil, badRPC(err)
+	}
+
+	// discovery should notice the connection event and do the dht thing
+	err = app.P2p.Host.Connect(app.Ctx, *info)
+
+	if err != nil {
+		return nil, badp2p(err)
+	}
+
+	return "addPeer success", nil
+}
+
+type beginAdvertisingMsg struct {
+}
+
+func (ap *beginAdvertisingMsg) run(app *app) (interface{}, error) {
+	if app.P2p == nil {
+		return nil, needsConfigure()
+	}
+
+	discovery.Advertise(app.Ctx, app.P2p.Discovery, app.P2p.Rendezvous)
+
+	return "beginAdvertising success", nil
+}
+
 var msgHandlers = map[methodIdx]func() action{
 	configure:           func() action { return &configureMsg{} },
 	listen:              func() action { return &listenMsg{} },
@@ -561,6 +629,8 @@ var msgHandlers = map[methodIdx]func() action{
 	removeStreamHandler: func() action { return &removeStreamHandlerMsg{} },
 	addStreamHandler:    func() action { return &addStreamHandlerMsg{} },
 	listeningAddrs:      func() action { return &listeningAddrsMsg{} },
+	addPeer:             func() action { return &addPeerMsg{} },
+	beginAdvertising:    func() action { return &beginAdvertisingMsg{} },
 }
 
 type errorResult struct {

src/libp2p_helper/methodidx_jsonenum.go

@@ -23,6 +23,8 @@ var (
 		"removeStreamHandler": removeStreamHandler,
 		"addStreamHandler":    addStreamHandler,
 		"listeningAddrs":      listeningAddrs,
+		"addPeer":             addPeer,
+		"beginAdvertising":    beginAdvertising,
 	}
 
 	_methodIdxValueToName = map[methodIdx]string{
@@ -40,6 +42,8 @@ var (
 		removeStreamHandler: "removeStreamHandler",
 		addStreamHandler:    "addStreamHandler",
 		listeningAddrs:      "listeningAddrs",
+		addPeer:             "addPeer",
+		beginAdvertising:    "beginAdvertising",
 	}
 )
 
@@ -61,6 +65,8 @@ func init() {
 			interface{}(removeStreamHandler).(fmt.Stringer).String(): removeStreamHandler,
 			interface{}(addStreamHandler).(fmt.Stringer).String():    addStreamHandler,
 			interface{}(listeningAddrs).(fmt.Stringer).String():      listeningAddrs,
+			interface{}(addPeer).(fmt.Stringer).String():             addPeer,
+			interface{}(beginAdvertising).(fmt.Stringer).String():    beginAdvertising,
 		}
 	}
 }