Merge pull request #522 from shutter-network/feature/251-floodsub-peer-discovery

floodsub discovery mechanism

Author: blockchainluffy

rolling-shutter/p2p/config.go

@@ -46,8 +46,9 @@ type Config struct {
 	AdvertiseAddresses       []*address.P2PAddress `comment:"Optional, addresses to be advertised to other peers instead of auto-detected ones."`
 	CustomBootstrapAddresses []*address.P2PAddress `comment:"Overwrite p2p boostrap nodes"`
 	Environment              env.Environment
-	DiscoveryNamespace       string `shconfig:",required" comment:"Must be unique for each instance id."`
-	IsAccessNode             bool   `comment:"Optional, to be set to true if running an access node"`
+	DiscoveryNamespace       string                  `shconfig:",required" comment:"Must be unique for each instance id."`
+	IsAccessNode             bool                    `comment:"Optional, to be set to true if running an access node"`
+	FloodSubDiscovery        FloodsubDiscoveryConfig `shconfig:"required"`
 }
 
 func (c *Config) Name() string {

rolling-shutter/p2p/floodsubpeerdiscovery/gossippeerdiscovery.go

@@ -0,0 +1,118 @@
+package floodsubpeerdiscovery
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/libp2p/go-libp2p/core/peerstore"
+	"github.com/rs/zerolog/log"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/address"
+)
+
+const defaultTopic = "_peer-discovery._p2p._pubsub"
+
+type FloodsubPeerDiscovery struct {
+	PeerDiscoveryComponents
+	Interval int
+	Topics   []*pubsub.Topic
+}
+
+type PeerDiscoveryComponents struct {
+	PeerID    address.P2PIdentifier
+	Pubsub    *pubsub.PubSub
+	PeerStore peerstore.Peerstore
+}
+
+func (pd *FloodsubPeerDiscovery) Init(config PeerDiscoveryComponents, interval int, topics []string) error {
+	pd.Interval = interval
+	pd.PeerID = config.PeerID
+	pd.Pubsub = config.Pubsub
+	pd.PeerStore = config.PeerStore
+
+	if len(topics) > 0 {
+		for _, topic := range topics {
+			topic, err := pd.Pubsub.Join(topic)
+			if err != nil {
+				return fmt.Errorf("failed to join topic | err %w", err)
+			}
+			pd.Topics = append(pd.Topics, topic)
+		}
+	} else {
+		topic, err := pd.Pubsub.Join(defaultTopic)
+		if err != nil {
+			return fmt.Errorf("failed to join topic | err %w", err)
+		}
+		pd.Topics = append(pd.Topics, topic)
+	}
+	return nil
+}
+
+func (pd *FloodsubPeerDiscovery) Start(ctx context.Context) error {
+	timer := time.NewTicker(time.Duration(pd.Interval) * time.Second)
+
+	for {
+		select {
+		case <-timer.C:
+			err := pd.broadcast()
+			if err != nil {
+				log.Warn().Msgf("error in broadcasting floodsub msg | %v", err)
+				return err
+			}
+		case <-ctx.Done():
+			for _, topic := range pd.Topics {
+				if err := topic.Close(); err != nil {
+					return fmt.Errorf("error in closing topic | %w", err)
+				}
+			}
+			return nil
+		}
+	}
+}
+
+func (pd *FloodsubPeerDiscovery) broadcast() error {
+	pubKey, err := pd.PeerID.ExtractPublicKey()
+	if err != nil {
+		return fmt.Errorf("peerId was missing public key | err %w", err)
+	}
+
+	pubKeyBytes, err := pubKey.Raw()
+	if err != nil || len(pubKeyBytes) == 0 {
+		return fmt.Errorf("peerId was missing public key | err %w", err)
+	}
+
+	if pd.Pubsub == nil {
+		return fmt.Errorf("pubSub not configured | err %w", err)
+	}
+
+	addresses := make([][]byte, 0)
+
+	for _, addr := range pd.PeerStore.Addrs(pd.PeerID.ID) {
+		addresses = append(addresses, addr.Bytes())
+	}
+
+	peer := Peer{
+		PublicKey: pubKeyBytes,
+		Addrs:     addresses,
+	}
+	pbPeer, err := proto.Marshal(&peer)
+	if err != nil {
+		return fmt.Errorf("error marshaling message | err %w", err)
+	}
+
+	for _, topic := range pd.Topics {
+		if len(pd.Pubsub.ListPeers(topic.String())) == 0 {
+			log.Info().Msgf("skipping broadcasting our peer data on topic %s because there are no peers present", topic)
+			continue
+		}
+		log.Info().Msgf("broadcasting our peer data on topic %s", topic)
+
+		if err := topic.Publish(context.Background(), pbPeer); err != nil {
+			return fmt.Errorf("failed to publish to topic | err %w", err)
+		}
+	}
+	return nil
+}

rolling-shutter/p2p/floodsubpeerdiscovery/peer.pb.go

@@ -0,0 +1,8 @@
+syntax = "proto3";
+
+option go_package = "./;floodsubpeerdiscovery";
+
+message Peer {
+  bytes publicKey = 1;
+  repeated bytes addrs = 2;
+}

rolling-shutter/p2p/floodsubpeerdiscovery/peer.proto

@@ -99,6 +99,7 @@ func New(config *Config) (*P2PMessaging, error) {
 		Environment:        config.Environment,
 		DiscoveryNamespace: config.DiscoveryNamespace,
 		IsAccessNode:       config.IsAccessNode,
+		FloodsubDiscovery:  config.FloodSubDiscovery,
 	}
 
 	bootstrapAddresses := config.CustomBootstrapAddresses

rolling-shutter/p2p/messaging.go

@@ -23,6 +23,7 @@ import (
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/env"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/keys"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/p2p/floodsubpeerdiscovery"
 )
 
 var DefaultBootstrapPeers []*address.P2PAddress
@@ -59,7 +60,8 @@ type P2PNode struct {
 	pubSub      *pubsub.PubSub
 	gossipRooms map[string]*gossipRoom
 
-	GossipMessages chan *pubsub.Message
+	GossipMessages    chan *pubsub.Message
+	FloodSubDiscovery *floodsubpeerdiscovery.FloodsubPeerDiscovery
 }
 
 type p2pNodeConfig struct {
@@ -71,6 +73,13 @@ type p2pNodeConfig struct {
 	IsBootstrapNode    bool
 	IsAccessNode       bool
 	DiscoveryNamespace string
+	FloodsubDiscovery  FloodsubDiscoveryConfig
+}
+
+type FloodsubDiscoveryConfig struct {
+	Enabled  bool
+	Interval int
+	Topics   []string
 }
 
 func NewP2PNode(config p2pNodeConfig) *P2PNode {
@@ -123,6 +132,22 @@ func (p *P2PNode) Run(
 			return room.readLoop(ctx, p.GossipMessages)
 		})
 	}
+	if p.config.FloodsubDiscovery.Enabled {
+		log.Info().Msg("floodsub peer discovery is enabled")
+		p.FloodSubDiscovery = &floodsubpeerdiscovery.FloodsubPeerDiscovery{}
+		peerDiscoveryComponents := floodsubpeerdiscovery.PeerDiscoveryComponents{
+			PeerID:    address.P2PIdentifier{ID: p.host.ID()},
+			PeerStore: p.host.Peerstore(),
+			Pubsub:    p.pubSub,
+		}
+		err := p.FloodSubDiscovery.Init(peerDiscoveryComponents, p.config.FloodsubDiscovery.Interval, p.config.FloodsubDiscovery.Topics)
+		if err != nil {
+			return err
+		}
+		runner.Go(func() error {
+			return p.FloodSubDiscovery.Start(ctx)
+		})
+	}
 	runner.Go(func() error {
 		log.Info().Str("namespace", p.config.DiscoveryNamespace).Msg("starting advertizing discovery node")
 		util.Advertise(ctx, p.discovery, p.config.DiscoveryNamespace)