From 70efcd72f3627a534c5abb97d09520698fa8fb86 Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Fri, 1 Oct 2021 13:49:50 -0400 Subject: [PATCH] feat: rendezvous (#54) --- go.mod | 1 + go.sum | 11 +++++++++++ waku/node.go | 31 ++++++++++++++++++++++++++++--- waku/v2/node/wakunode2.go | 31 +++++++++++++++++++++++++++++-- waku/v2/node/wakuoptions.go | 22 ++++++++++++++++++++++ 5 files changed, 91 insertions(+), 5 deletions(-) diff --git a/go.mod b/go.mod index 6cfa9e1a..febc9330 100644 --- a/go.mod +++ b/go.mod @@ -30,6 +30,7 @@ require ( github.com/spf13/cobra v1.1.3 github.com/spf13/jwalterweatherman v1.1.0 // indirect github.com/spf13/viper v1.7.1 + github.com/status-im/go-libp2p-rendezvous v0.0.0-20210928230014-94a02b1432a3 // indirect github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20210729162817-adc68830282a github.com/stretchr/testify v1.7.0 go.opencensus.io v0.23.0 diff --git a/go.sum b/go.sum index 80bdae14..d91e3d59 100644 --- a/go.sum +++ b/go.sum @@ -238,6 +238,8 @@ github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zV github.com/gogo/protobuf v1.3.0/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o= github.com/gogo/protobuf v1.3.1 h1:DqDEcV5aeaTmdFBePNpYsp3FlcVH/2ISVVM9Qf8PSls= github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k= github.com/golang/geo v0.0.0-20190916061304-5b978397cfec/go.mod h1:QZ0nwyI2jOfgRAoBvP+ab5aRr7c9x7lhGEJrKvBwjWI= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= @@ -459,6 +461,7 @@ github.com/kami-zh/go-capturer v0.0.0-20171211120116-e492ea43421d/go.mod h1:P2vi github.com/karalabe/usb v0.0.0-20190919080040-51dc0efba356/go.mod h1:Od972xHfMJowv7NGVDiWVxk2zxnWgjLlJzE+F4F7AGU= github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q= github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4= github.com/klauspost/compress v1.4.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A= @@ -942,6 +945,12 @@ github.com/spf13/viper v1.7.1/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5q github.com/src-d/envconfig v1.0.0/go.mod h1:Q9YQZ7BKITldTBnoxsE5gOeB5y66RyPXeue/R4aaNBc= github.com/status-im/go-ethereum v1.10.4-status.2 h1:uvcD2U7skYqPQviARFb4w3wZyFSYLs/pfVrJaRSDcCA= github.com/status-im/go-ethereum v1.10.4-status.2/go.mod h1:GvIhpdCOgMHI6i5xVPEZOrv/qSMeOFHbZh77AoyZUoE= +github.com/status-im/go-libp2p-rendezvous v0.0.0-20210928195017-4397dd0f844d h1:cJkS8VzW0jwOne1ipKiUCjMLbMQU7Xa7XPJCMrjZCNU= +github.com/status-im/go-libp2p-rendezvous v0.0.0-20210928195017-4397dd0f844d/go.mod h1:up+uGvIr4JbhszgZB4fLcFoc0xbihgv1M885ZQrgtIs= +github.com/status-im/go-libp2p-rendezvous v0.0.0-20210928205005-be9b1a0035bf h1:hhkIKjPCMuDd1T5h0DPc4dS2HUc1tzkMw1ReWj7stnY= +github.com/status-im/go-libp2p-rendezvous v0.0.0-20210928205005-be9b1a0035bf/go.mod h1:up+uGvIr4JbhszgZB4fLcFoc0xbihgv1M885ZQrgtIs= +github.com/status-im/go-libp2p-rendezvous v0.0.0-20210928230014-94a02b1432a3 h1:C6Ed5GA9wlvNOZeLi16hMVpEXLb+4AhGSDx/ZljtWbs= +github.com/status-im/go-libp2p-rendezvous v0.0.0-20210928230014-94a02b1432a3/go.mod h1:up+uGvIr4JbhszgZB4fLcFoc0xbihgv1M885ZQrgtIs= github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20210729162817-adc68830282a h1:eCna/q/PuZVqtmOMBqytw9yzZwMNKpao4au0OJDvesI= github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20210729162817-adc68830282a/go.mod h1:LSCVYR7mnBBsxVJghrGpQ3yJAAATEe6XeQQqGCZhwrE= github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4/go.mod h1:RZLeN1LMWmRsyYjvAu+I6Dm9QmlDaIIt+Y+4Kd7Tp+Q= @@ -1280,9 +1289,11 @@ golang.org/x/tools v0.0.0-20200501065659-ab2804fb9c9d/go.mod h1:EkVYQZoAsY45+roY golang.org/x/tools v0.0.0-20200512131952-2bc93b1c0c88/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200515010526-7d3b6ebf133d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200618134242-20370b0cb4b2/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.0 h1:po9/4sTYwZU9lPhi1tOrb4hCv3qrhiQ77LZfGa2OjwY= golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/waku/node.go b/waku/node.go index 5222fb61..0d2b7751 100644 --- a/waku/node.go +++ b/waku/node.go @@ -18,6 +18,7 @@ import ( dssql "github.com/ipfs/go-ds-sql" logging "github.com/ipfs/go-log" "github.com/libp2p/go-libp2p" + "github.com/libp2p/go-libp2p-core/peer" "github.com/libp2p/go-libp2p-peerstore/pstoreds" "github.com/multiformats/go-multiaddr" "github.com/spf13/cobra" @@ -25,12 +26,13 @@ import ( "github.com/status-im/go-waku/waku/metrics" "github.com/status-im/go-waku/waku/persistence" "github.com/status-im/go-waku/waku/persistence/sqlite" + pubsub "github.com/status-im/go-wakurelay-pubsub" "github.com/status-im/go-waku/waku/v2/discovery" "github.com/status-im/go-waku/waku/v2/node" "github.com/status-im/go-waku/waku/v2/protocol/relay" - pubsub "github.com/status-im/go-wakurelay-pubsub" + libp2pdisc "github.com/libp2p/go-libp2p-core/discovery" ) var log = logging.Logger("wakunode") @@ -82,6 +84,9 @@ var rootCmd = &cobra.Command{ dnsDiscoveryUrl, _ := cmd.Flags().GetString("dns-discovery-url") dnsDiscoveryNameServer, _ := cmd.Flags().GetString("dns-discovery-nameserver") peerExchange, _ := cmd.Flags().GetBool("peer-exchange") + enableRendezvous, _ := cmd.Flags().GetBool("rendezvous") + rendezvousPeerIds, _ := cmd.Flags().GetStringSlice("rendezvous-nodes") + enableRendezvousServer, _ := cmd.Flags().GetBool("rendezvous-server") hostAddr, _ := net.ResolveTCPAddr("tcp", fmt.Sprint("0.0.0.0:", port)) @@ -152,6 +157,22 @@ var rootCmd = &cobra.Command{ nodeOpts = append(nodeOpts, node.WithWakuRelay(wakurelayopts...)) } + if enableRendezvous && len(rendezvousPeerIds) > 0 { + var peers []peer.ID + for _, r := range rendezvousPeerIds { + peerId, err := peer.Decode(r) + if err != nil { + checkError(err, "Rendezvous") + } + peers = append(peers, peerId) + } + nodeOpts = append(nodeOpts, node.WithRendezvous(peers, pubsub.WithDiscoveryOpts(libp2pdisc.TTL(time.Duration(20)*time.Second)))) + } + + if enableRendezvousServer { + nodeOpts = append(nodeOpts, node.WithRendezvousServer()) + } + if wakuFilter { nodeOpts = append(nodeOpts, node.WithWakuFilter()) } @@ -198,13 +219,14 @@ var rootCmd = &cobra.Command{ ctx, cancel := context.WithTimeout(ctx, time.Duration(3)*time.Second) defer cancel() err = wakuNode.DialPeer(ctx, node) - checkError(err, "error dialing peer") + if err != nil { + log.Error("error dialing peer ", err) + } }(n) } } if enableDnsDiscovery { - for _, addr := range wakuNode.ListenAddresses() { ip, _ := addr.ValueForProtocol(multiaddr.P_IP4) enr := enode.NewV4(&prvKey.PublicKey, net.ParseIP(ip), hostAddr.Port, 0) @@ -312,6 +334,9 @@ func init() { rootCmd.Flags().String("dns-discovery-url", "", "URL for DNS node list in format 'enrtree://@'") rootCmd.Flags().String("dns-discovery-nameserver", "", "DNS nameserver IP to query (empty to use system's default)") rootCmd.Flags().Bool("peer-exchange", true, "Enable GossipSub Peer Exchange") + rootCmd.Flags().Bool("rendezvous", false, "Enable rendezvous for peer discovery") + rootCmd.Flags().StringSlice("rendezvous-nodes", []string{}, "Peer IDs of waku2 rendezvous nodes. Argument may be repeated") + rootCmd.Flags().Bool("rendezvous-server", false, "Node will act as rendezvous server") } func initConfig() { diff --git a/waku/v2/node/wakunode2.go b/waku/v2/node/wakunode2.go index 3f3a0ebe..f64b4866 100644 --- a/waku/v2/node/wakunode2.go +++ b/waku/v2/node/wakunode2.go @@ -26,6 +26,7 @@ import ( "go.opencensus.io/stats" "go.opencensus.io/tag" + rendezvous "github.com/status-im/go-libp2p-rendezvous" "github.com/status-im/go-waku/waku/v2/metrics" "github.com/status-im/go-waku/waku/v2/protocol" "github.com/status-im/go-waku/waku/v2/protocol/filter" @@ -34,6 +35,8 @@ import ( "github.com/status-im/go-waku/waku/v2/protocol/relay" "github.com/status-im/go-waku/waku/v2/protocol/store" wakurelay "github.com/status-im/go-wakurelay-pubsub" + + db "github.com/status-im/go-libp2p-rendezvous/db/sqlite" ) var log = logging.Logger("wakunode") @@ -60,6 +63,8 @@ type WakuNode struct { filter *filter.WakuFilter lightPush *lightpush.WakuLightPush + rendezvous *rendezvous.RendezvousService + ping *ping.PingService subscriptions map[relay.Topic][]*Subscription @@ -197,13 +202,13 @@ func New(ctx context.Context, opts ...WakuNodeOption) (*WakuNode, error) { params := new(WakuNodeParameters) ctx, cancel := context.WithCancel(ctx) - _ = cancel params.libP2POpts = DefaultLibP2POptions for _, opt := range opts { err := opt(params) if err != nil { + cancel() return nil, err } } @@ -218,6 +223,7 @@ func New(ctx context.Context, opts ...WakuNodeOption) (*WakuNode, error) { host, err := libp2p.New(ctx, params.libP2POpts...) if err != nil { + cancel() return nil, err } @@ -260,6 +266,11 @@ func New(ctx context.Context, opts ...WakuNodeOption) (*WakuNode, error) { } } + if params.enableRendezvous { + rendezvous := rendezvous.NewRendezvousDiscovery(w.host, params.rendezvousPeers) + params.wOpts = append(params.wOpts, wakurelay.WithDiscovery(rendezvous, params.rendezvousOpts...)) + } + err = w.mountRelay(params.enableRelay, params.wOpts...) if err != nil { return nil, err @@ -273,6 +284,13 @@ func New(ctx context.Context, opts ...WakuNodeOption) (*WakuNode, error) { w.startKeepAlive(params.keepAliveInterval) } + if params.enableRendezvousServer { + err := w.mountRendezvous() + if err != nil { + return nil, err + } + } + for _, addr := range w.ListenAddresses() { log.Info("Listening on ", addr) } @@ -402,6 +420,16 @@ func (w *WakuNode) mountLightPush() { w.lightPush = lightpush.NewWakuLightPush(w.ctx, w.host, w.relay) } +func (w *WakuNode) mountRendezvous() error { + dbi, err := db.OpenDB(w.ctx, ":memory:") // TODO: replace for levelDB + if err != nil { + return err + } + w.rendezvous = rendezvous.NewRendezvousService(w.host, dbi) + log.Info("Rendezvous service started") + return nil +} + func (w *WakuNode) AddPeer(info *peer.AddrInfo, protocolId string) error { log.Info(fmt.Sprintf("adding peer %s with protocol %s", info.ID.Pretty(), protocolId)) @@ -785,7 +813,6 @@ func (w *WakuNode) startKeepAlive(t time.Duration) { w.ping = ping.NewPingService(w.host) ticker := time.NewTicker(t) go func() { - // This map contains peers that we're // waiting for the ping response from peerMap := make(map[peer.ID]<-chan ping.Result) diff --git a/waku/v2/node/wakuoptions.go b/waku/v2/node/wakuoptions.go index 13c5eecc..d0e8babc 100644 --- a/waku/v2/node/wakuoptions.go +++ b/waku/v2/node/wakuoptions.go @@ -8,6 +8,7 @@ import ( "github.com/libp2p/go-libp2p" connmgr "github.com/libp2p/go-libp2p-connmgr" "github.com/libp2p/go-libp2p-core/crypto" + "github.com/libp2p/go-libp2p-core/peer" ma "github.com/multiformats/go-multiaddr" manet "github.com/multiformats/go-multiaddr-net" "github.com/status-im/go-waku/waku/v2/protocol/store" @@ -32,6 +33,11 @@ type WakuNodeParameters struct { store *store.WakuStore // filter *filter.WakuFilter + enableRendezvous bool + enableRendezvousServer bool + rendezvousPeers []peer.ID + rendezvousOpts []wakurelay.DiscoverOpt + keepAliveInterval time.Duration enableLightPush bool @@ -96,6 +102,22 @@ func WithWakuRelay(opts ...wakurelay.Option) WakuNodeOption { } } +func WithRendezvous(peers []peer.ID, discoverOpts ...wakurelay.DiscoverOpt) WakuNodeOption { + return func(params *WakuNodeParameters) error { + params.enableRendezvous = true + params.rendezvousPeers = peers + params.rendezvousOpts = discoverOpts + return nil + } +} + +func WithRendezvousServer() WakuNodeOption { + return func(params *WakuNodeParameters) error { + params.enableRendezvousServer = true + return nil + } +} + // WithWakuFilter enables the Waku V2 Filter protocol. func WithWakuFilter() WakuNodeOption { return func(params *WakuNodeParameters) error {