feat: implemented and tested new floodsub discovery messaging

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/gossippeerdiscovery/gossippeerdiscovery.go renamed to rolling-shutter/p2p/floodsubpeerdiscovery/gossippeerdiscovery.go

@@ -1,8 +1,9 @@
-package gossippeerdiscovery
+package floodsubpeerdiscovery
 
 import (
 	"context"
 	"fmt"
+	"time"
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/peerstore"
@@ -13,34 +14,61 @@ import (
 
 const defaultTopic = "_peer-discovery._p2p._pubsub"
 
-type GossipPeerDiscovery struct {
+type FloodsubPeerDiscovery struct {
 	PeerDiscoveryComponents
-	Interval   int
-	Topics     []string
-	ListenOnly bool
+	Interval int
+	Topics   []*pubsub.Topic
 }
 
 type PeerDiscoveryComponents struct {
-	peerId    address.P2PIdentifier
-	pubsub    *pubsub.PubSub
-	peerStore peerstore.Peerstore
+	PeerId    address.P2PIdentifier
+	Pubsub    *pubsub.PubSub
+	PeerStore peerstore.Peerstore
 }
 
-func (pd *GossipPeerDiscovery) init(config PeerDiscoveryComponents, interval int, topics []string, listenOnly bool) {
+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 {
-		pd.Topics = topics
+		for _, topic := range topics {
+			topic, err := pd.Pubsub.Join(topic)
+			if err != nil {
+				return fmt.Errorf("failed to join topic | err %v", err)
+			}
+			pd.Topics = append(pd.Topics, topic)
+		}
 	} else {
-		pd.Topics = []string{defaultTopic}
+		topic, err := pd.Pubsub.Join(defaultTopic)
+		if err != nil {
+			return fmt.Errorf("failed to join topic | err %v", err)
+		}
+		pd.Topics = append(pd.Topics, topic)
 	}
-	pd.ListenOnly = listenOnly
-	pd.peerId = config.peerId
-	pd.pubsub = config.pubsub
-	pd.peerStore = config.peerStore
+	return nil
 }
 
-func (pd *GossipPeerDiscovery) broadcast() error {
-	pubKey, err := pd.peerId.ExtractPublicKey()
+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.Info().Msgf("error in broadcasting floodsub msg | %v", err)
+				return err
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+func (pd *FloodsubPeerDiscovery) broadcast() error {
+	pubKey, err := pd.PeerId.ExtractPublicKey()
 	if err != nil {
 		return fmt.Errorf("peerId was missing public key | err %v", err)
 	}
@@ -50,13 +78,13 @@ func (pd *GossipPeerDiscovery) broadcast() error {
 		return fmt.Errorf("peerId was missing public key | err %v", err)
 	}
 
-	if pd.pubsub == nil {
+	if pd.Pubsub == nil {
 		return fmt.Errorf("pubSub not configured | err %v", err)
 	}
 
 	addresses := make([][]byte, 0)
 
-	for _, addr := range pd.peerStore.Addrs(pd.peerId.ID) {
+	for _, addr := range pd.PeerStore.Addrs(pd.PeerId.ID) {
 		addresses = append(addresses, addr.Bytes())
 	}
 
@@ -70,18 +98,16 @@ func (pd *GossipPeerDiscovery) broadcast() error {
 	}
 
 	for _, topic := range pd.Topics {
-		if len(pd.pubsub.ListPeers(topic)) == 0 {
+		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)
-		topic, err := pd.pubsub.Join(topic)
-		if err != nil {
-			return fmt.Errorf("failed to join topic | err %v", err)
-		}
+
 		if err := topic.Publish(context.Background(), pbPeer); err != nil {
 			return fmt.Errorf("failed to publish to topic | err %v", err)
 		}
+		defer topic.Close()
 	}
 	return nil
 }

rolling-shutter/p2p/gossippeerdiscovery/peer.pb.go renamed to rolling-shutter/p2p/floodsubpeerdiscovery/peer.pb.go

@@ -1,6 +1,6 @@
 syntax = "proto3";
 
-option go_package = "./;gossippeerdiscovery";
+option go_package = "./;floodsubpeerdiscovery";
 
 message Peer {
   bytes publicKey = 1;

rolling-shutter/p2p/gossippeerdiscovery/peer.proto renamed to 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,19 @@ 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{p.host.ID()},
+			PeerStore: p.host.Peerstore(),
+			Pubsub:    p.pubSub,
+		}
+		p.FloodSubDiscovery.Init(peerDiscoveryComponents, p.config.FloodsubDiscovery.Interval, p.config.FloodsubDiscovery.Topics)
+		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)