Merge branch 'master' into storenode-cycle

library/c/README.md

@@ -285,6 +285,8 @@ interface JsonConfig {
     storeRetentionTimeSeconds?: number;
     websocket?: Websocket;
     dns4DomainName?: string;
+    clusterID: int;
+    shards: Array<uint16>;
 }
 ```
 

library/config.go

@@ -31,6 +31,8 @@ type WakuConfig struct {
 	RetentionTimeSeconds   *int             `json:"storeRetentionTimeSeconds,omitempty"`
 	DNS4DomainName         string           `json:"dns4DomainName,omitempty"`
 	Websockets             *WebsocketConfig `json:"websockets,omitempty"`
+	ClusterID              int              `json:"clusterID"`
+	Shards                 []uint16         `json:"shards"`
 }
 
 // WebsocketConfig contains all the settings required to setup websocket support in waku

library/node.go

@@ -164,6 +164,8 @@ func NewNode(instance *WakuInstance, configJSON string) error {
 		node.WithPrivateKey(prvKey),
 		node.WithHostAddress(hostAddr),
 		node.WithKeepAlive(10*time.Second, time.Duration(*config.KeepAliveInterval)*time.Second),
+		node.WithClusterID(uint16(config.ClusterID)),
+		node.WithShards(config.Shards),
 	}
 
 	if *config.EnableRelay {

waku/v2/api/filter/filter.go

@@ -3,10 +3,12 @@ package filter
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/p2p/net/swarm"
 	"github.com/waku-org/go-waku/waku/v2/onlinechecker"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/filter"
@@ -29,6 +31,7 @@ func (fc FilterConfig) String() string {
 }
 
 const filterSubLoopInterval = 5 * time.Second
+const filterSubMaxErrCnt = 3
 
 type Sub struct {
 	ContentFilter         protocol.ContentFilter
@@ -43,6 +46,7 @@ type Sub struct {
 	onlineChecker         onlinechecker.OnlineChecker
 	resubscribeInProgress bool
 	id                    string
+	errcnt                int
 }
 
 type subscribeParameters struct {
@@ -107,13 +111,14 @@ func (apiSub *Sub) Unsubscribe(contentFilter protocol.ContentFilter) {
 	}
 }
 
-func (apiSub *Sub) subscriptionLoop(batchInterval time.Duration) {
+func (apiSub *Sub) subscriptionLoop(loopInterval time.Duration) {
 	defer utils.LogOnPanic()
-	ticker := time.NewTicker(batchInterval)
+	ticker := time.NewTicker(loopInterval)
 	defer ticker.Stop()
 	for {
 		select {
 		case <-ticker.C:
+			apiSub.errcnt = 0 //reset errorCount
 			if apiSub.onlineChecker.IsOnline() && len(apiSub.subs) < apiSub.Config.MaxPeers &&
 				!apiSub.resubscribeInProgress && len(apiSub.closing) < apiSub.Config.MaxPeers {
 				apiSub.closing <- ""
@@ -123,9 +128,11 @@ func (apiSub *Sub) subscriptionLoop(batchInterval time.Duration) {
 			apiSub.cleanup()
 			return
 		case subId := <-apiSub.closing:
-			apiSub.resubscribeInProgress = true
-			//trigger resubscribe flow for subscription.
-			apiSub.checkAndResubscribe(subId)
+			if apiSub.errcnt < filterSubMaxErrCnt {
+				apiSub.resubscribeInProgress = true
+				//trigger resubscribe flow for subscription.
+				apiSub.checkAndResubscribe(subId)
+			}
 		}
 	}
 }
@@ -181,6 +188,10 @@ func (apiSub *Sub) resubscribe(failedPeer peer.ID) {
 	apiSub.multiplex(subs)
 }
 
+func possibleRecursiveError(err error) bool {
+	return errors.Is(err, utils.ErrNoPeersAvailable) || errors.Is(err, swarm.ErrDialBackoff)
+}
+
 func (apiSub *Sub) subscribe(contentFilter protocol.ContentFilter, peerCount int, peersToExclude ...peer.ID) ([]*subscription.SubscriptionDetails, error) {
 	// Low-level subscribe, returns a set of SubscriptionDetails
 	options := make([]filter.FilterSubscribeOption, 0)
@@ -195,6 +206,9 @@ func (apiSub *Sub) subscribe(contentFilter protocol.ContentFilter, peerCount int
 	subs, err := apiSub.wf.Subscribe(apiSub.ctx, contentFilter, options...)
 
 	if err != nil {
+		if possibleRecursiveError(err) {
+			apiSub.errcnt++
+		}
 		//Inform of error, so that resubscribe can be triggered if required
 		if len(apiSub.closing) < apiSub.Config.MaxPeers {
 			apiSub.closing <- ""

waku/v2/discv5/discover.go

@@ -21,7 +21,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/ethereum/go-ethereum/p2p/enode"
-	"github.com/ethereum/go-ethereum/p2p/enr"
 	"github.com/ethereum/go-ethereum/p2p/nat"
 )
 
@@ -255,19 +254,13 @@ func (d *DiscoveryV5) Stop() {
 }
 
 func isWakuNode(node *enode.Node) bool {
-	enrField := new(wenr.WakuEnrBitfield)
-	if err := node.Record().Load(enr.WithEntry(wenr.WakuENRField, &enrField)); err != nil {
-		if !enr.IsNotFound(err) {
-			utils.Logger().Named("discv5").Error("could not retrieve waku2 ENR field for enr ", zap.Any("node", node))
-		}
+	enrField, err := wenr.GetWakuEnrBitField(node)
+	if err != nil {
+		utils.Logger().Named("discv5").Error("could not retrieve waku2 ENR field for enr ", zap.Error(err))
 		return false
 	}
 
-	if enrField != nil {
-		return *enrField != uint8(0) // #RFC 31 requirement
-	}
-
-	return false
+	return enrField != uint8(0) // #RFC 31 requirement
 }
 
 func (d *DiscoveryV5) evaluateNode() func(node *enode.Node) bool {

waku/v2/discv5/filters.go

@@ -4,7 +4,6 @@ import (
 	wenr "github.com/waku-org/go-waku/waku/v2/protocol/enr"
 
 	"github.com/ethereum/go-ethereum/p2p/enode"
-	"github.com/ethereum/go-ethereum/p2p/enr"
 )
 
 // FilterPredicate is to create a Predicate using a custom function
@@ -36,16 +35,11 @@ func FilterShard(cluster, index uint16) Predicate {
 func FilterCapabilities(flags wenr.WakuEnrBitfield) Predicate {
 	return func(iterator enode.Iterator) enode.Iterator {
 		predicate := func(node *enode.Node) bool {
-			enrField := new(wenr.WakuEnrBitfield)
-			if err := node.Record().Load(enr.WithEntry(wenr.WakuENRField, &enrField)); err != nil {
+			enrField, err := wenr.GetWakuEnrBitField(node)
+			if err != nil {
 				return false
 			}
-
-			if enrField == nil {
-				return false
-			}
-
-			return *enrField&flags == flags
+			return enrField&flags == flags
 		}
 		return enode.Filter(iterator, predicate)
 	}

waku/v2/node/metrics.go

@@ -3,7 +3,6 @@ package node
 import (
 	"fmt"
 
-	"github.com/libp2p/go-libp2p/core/metrics"
 	"github.com/libp2p/go-libp2p/p2p/metricshelper"
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -34,20 +33,11 @@ var peerStoreSize = prometheus.NewGauge(
 		Help: "Size of Peer Store",
 	})
 
-var bandwidthTotal = prometheus.NewCounterVec(
-	prometheus.CounterOpts{
-		Name: "libp2p_network_bytes_total",
-		Help: "Bandwidth usage total",
-	},
-	[]string{"direction"},
-)
-
 var collectors = []prometheus.Collector{
 	gitVersion,
 	peerDials,
 	connectedPeers,
 	peerStoreSize,
-	bandwidthTotal,
 }
 
 // Metrics exposes the functions required to update prometheus metrics for the waku node
@@ -57,7 +47,6 @@ type Metrics interface {
 	RecordPeerConnected()
 	RecordPeerDisconnected()
 	SetPeerStoreSize(int)
-	RecordBandwidth(metrics.Stats)
 }
 
 type metricsImpl struct {
@@ -95,9 +84,3 @@ func (m *metricsImpl) RecordPeerDisconnected() {
 func (m *metricsImpl) SetPeerStoreSize(size int) {
 	peerStoreSize.Set(float64(size))
 }
-
-func (m *metricsImpl) RecordBandwidth(stats metrics.Stats) {
-	bandwidthTotal.WithLabelValues("in").Add(float64(stats.TotalIn))
-	bandwidthTotal.WithLabelValues("out").Add(float64(stats.TotalOut))
-
-}

waku/v2/node/wakunode2.go

@@ -18,7 +18,6 @@ import (
 
 	"github.com/libp2p/go-libp2p/core/event"
 	"github.com/libp2p/go-libp2p/core/host"
-	"github.com/libp2p/go-libp2p/core/metrics"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/peerstore"
@@ -85,12 +84,11 @@ type RLNRelay interface {
 }
 
 type WakuNode struct {
-	host             host.Host
-	opts             *WakuNodeParameters
-	log              *zap.Logger
-	timesource       timesource.Timesource
-	metrics          Metrics
-	bandwidthCounter *metrics.BandwidthCounter
+	host       host.Host
+	opts       *WakuNodeParameters
+	log        *zap.Logger
+	timesource timesource.Timesource
+	metrics    Metrics
 
 	peerstore     peerstore.Peerstore
 	peerConnector *peermanager.PeerConnectionStrategy
@@ -197,9 +195,6 @@ func New(opts ...WakuNodeOption) (*WakuNode, error) {
 	w.metrics = newMetrics(params.prometheusReg)
 	w.metrics.RecordVersion(Version, GitCommit)
 
-	w.bandwidthCounter = metrics.NewBandwidthCounter()
-	params.libP2POpts = append(params.libP2POpts, libp2p.BandwidthReporter(w.bandwidthCounter))
-
 	// Setup peerstore wrapper
 	if params.peerstore != nil {
 		w.peerstore = wps.NewWakuPeerstore(params.peerstore)
@@ -364,23 +359,6 @@ func (w *WakuNode) Start(ctx context.Context) error {
 
 	w.host = host
 
-	// Bandwidth reporter created for comparing IDONTWANT performance
-	go func() {
-		ticker := time.NewTicker(time.Second)
-		defer ticker.Stop()
-
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
-				totals := w.bandwidthCounter.GetBandwidthTotals()
-				w.bandwidthCounter.Reset()
-				w.metrics.RecordBandwidth(totals)
-			}
-		}
-	}()
-
 	if w.addressChangesSub, err = host.EventBus().Subscribe(new(event.EvtLocalAddressesUpdated)); err != nil {
 		return err
 	}

waku/v2/node/wakuoptions.go

@@ -319,6 +319,11 @@ func WithPrivateKey(privKey *ecdsa.PrivateKey) WakuNodeOption {
 func WithClusterID(clusterID uint16) WakuNodeOption {
 	return func(params *WakuNodeParameters) error {
 		params.clusterID = clusterID
+		if params.shards == nil {
+			var pshards protocol.RelayShards
+			pshards.ClusterID = params.clusterID
+			params.shards = &pshards
+		}
 		return nil
 	}
 }
@@ -340,6 +345,18 @@ func WithPubSubTopics(topics []string) WakuNodeOption {
 	}
 }
 
+func WithShards(shards []uint16) WakuNodeOption {
+	return func(params *WakuNodeParameters) error {
+		if params.shards == nil {
+			var pshards protocol.RelayShards
+			pshards.ClusterID = params.clusterID
+			params.shards = &pshards
+		}
+		params.shards.ShardIDs = shards
+		return nil
+	}
+}
+
 // WithMaxConnectionsPerIP sets the max number of allowed peers from the same IP
 func WithMaxConnectionsPerIP(limit int) WakuNodeOption {
 	return func(params *WakuNodeParameters) error {

waku/v2/peermanager/fastest_peer_selector.go

@@ -139,7 +139,7 @@ func (r *FastestPeerSelector) FastestPeer(ctx context.Context, peers peer.IDSlic
 		}
 	}
 
-	return "", ErrNoPeersAvailable
+	return "", utils.ErrNoPeersAvailable
 }
 
 type pingResult struct {

waku/v2/peermanager/fastest_peer_selector_test.go

@@ -34,13 +34,13 @@ func TestRTT(t *testing.T) {
 	h3.Close()
 
 	_, err = rtt.FastestPeer(ctx, peer.IDSlice{h3.ID()})
-	require.ErrorIs(t, err, ErrNoPeersAvailable)
+	require.ErrorIs(t, err, utils.ErrNoPeersAvailable)
 
 	// H3 should never return
 	for i := 0; i < 100; i++ {
 		p, err := rtt.FastestPeer(ctx, peer.IDSlice{h2.ID(), h3.ID()})
 		if err != nil {
-			require.ErrorIs(t, err, ErrNoPeersAvailable)
+			require.ErrorIs(t, err, utils.ErrNoPeersAvailable)
 		} else {
 			require.NotEqual(t, h3.ID(), p)
 		}

waku/v2/peermanager/peer_manager.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ethereum/go-ethereum/p2p/enr"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/event"
 	"github.com/libp2p/go-libp2p/core/host"
@@ -99,10 +98,6 @@ const (
 	LowestRTT
 )
 
-// ErrNoPeersAvailable is emitted when no suitable peers are found for
-// some protocol
-var ErrNoPeersAvailable = errors.New("no suitable peers found")
-
 const maxFailedAttempts = 5
 const prunePeerStoreInterval = 10 * time.Minute
 const peerConnectivityLoopSecs = 15
@@ -547,8 +542,8 @@ func (pm *PeerManager) processPeerENR(p *service.PeerData) []protocol.ID {
 	}
 	supportedProtos := []protocol.ID{}
 	//Identify and specify protocols supported by the peer based on the discovered peer's ENR
-	var enrField wenr.WakuEnrBitfield
-	if err := p.ENR.Record().Load(enr.WithEntry(wenr.WakuENRField, &enrField)); err == nil {
+	enrField, err := wenr.GetWakuEnrBitField(p.ENR)
+	if err == nil {
 		for proto, protoENR := range pm.wakuprotoToENRFieldMap {
 			protoENRField := protoENR.waku2ENRBitField
 			if protoENRField&enrField != 0 {

waku/v2/peermanager/peer_manager_test.go

@@ -93,7 +93,7 @@ func TestServiceSlots(t *testing.T) {
 	defer h4.Close()
 
 	_, err = pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: protocol1})
-	require.Error(t, err, ErrNoPeersAvailable)
+	require.Error(t, err, utils.ErrNoPeersAvailable)
 
 	// add h4 peer for protocol1
 	_, err = pm.AddPeer(tests.GetAddr(h4), wps.Static, []string{""}, libp2pProtocol.ID(protocol1))
@@ -138,7 +138,7 @@ func TestPeerSelection(t *testing.T) {
 	require.Equal(t, h2.ID(), peerIDs[0])
 
 	_, err = pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: protocol, PubsubTopics: []string{"/waku/2/rs/2/3"}})
-	require.Error(t, ErrNoPeersAvailable, err)
+	require.Error(t, utils.ErrNoPeersAvailable, err)
 
 	_, err = pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: protocol, PubsubTopics: []string{"/waku/2/rs/2/1"}})
 	require.NoError(t, err)
@@ -175,7 +175,7 @@ func TestDefaultProtocol(t *testing.T) {
 	///////////////
 	//Test empty peer selection for relay protocol
 	_, err := pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: relay.WakuRelayID_v200})
-	require.Error(t, err, ErrNoPeersAvailable)
+	require.Error(t, err, utils.ErrNoPeersAvailable)
 
 	///////////////
 	// getting peer for default protocol
@@ -215,7 +215,7 @@ func TestAdditionAndRemovalOfPeer(t *testing.T) {
 
 	pm.RemovePeer(peers[0])
 	_, err = pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: protocol2})
-	require.Error(t, err, ErrNoPeersAvailable)
+	require.Error(t, err, utils.ErrNoPeersAvailable)
 }
 
 func TestConnectToRelayPeers(t *testing.T) {

waku/v2/peermanager/peer_selection.go

@@ -9,6 +9,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/protocol"
 	wps "github.com/waku-org/go-waku/waku/v2/peerstore"
 	waku_proto "github.com/waku-org/go-waku/waku/v2/protocol"
+	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
 	"golang.org/x/exp/maps"
 )
@@ -59,7 +60,7 @@ func (pm *PeerManager) SelectRandom(criteria PeerSelectionCriteria) (peer.IDSlic
 	peerIDs, err := pm.selectServicePeer(criteria)
 	if err == nil && len(peerIDs) == criteria.MaxPeers {
 		return maps.Keys(peerIDs), nil
-	} else if !errors.Is(err, ErrNoPeersAvailable) {
+	} else if !errors.Is(err, utils.ErrNoPeersAvailable) {
 		pm.logger.Debug("could not retrieve random peer from slot", zap.String("protocol", string(criteria.Proto)),
 			zap.Strings("pubsubTopics", criteria.PubsubTopics), zap.Error(err))
 		return nil, err
@@ -101,7 +102,7 @@ func getRandom(filter PeerSet, count int, excludePeers PeerSet) (PeerSet, error)
 		}
 	}
 	if len(selectedPeers) == 0 {
-		return nil, ErrNoPeersAvailable
+		return nil, utils.ErrNoPeersAvailable
 	}
 	return selectedPeers, nil
 }
@@ -157,7 +158,7 @@ func (pm *PeerManager) selectServicePeer(criteria PeerSelectionCriteria) (PeerSe
 	if len(peers) == 0 {
 		pm.logger.Debug("could not retrieve random peer from slot", zap.Error(err))
 	}
-	return peers, ErrNoPeersAvailable
+	return peers, utils.ErrNoPeersAvailable
 }
 
 // PeerSelectionCriteria is the selection Criteria that is used by PeerManager to select peers.

waku/v2/peermanager/service_slot_test.go

@@ -6,6 +6,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 	libp2pProtocol "github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/stretchr/testify/require"
+	"github.com/waku-org/go-waku/waku/v2/utils"
 	"golang.org/x/exp/maps"
 )
 
@@ -26,7 +27,7 @@ func TestServiceSlot(t *testing.T) {
 	slots.getPeers(protocol).remove(peerID)
 	//
 	_, err = slots.getPeers(protocol).getRandom(1, nil)
-	require.Equal(t, err, ErrNoPeersAvailable)
+	require.Equal(t, err, utils.ErrNoPeersAvailable)
 
 	// Test with more peers
 	peerID2 := peer.ID("peerId2")
@@ -74,7 +75,7 @@ func TestServiceSlotRemovePeerFromAll(t *testing.T) {
 	slots.removePeer(peerID)
 	//
 	_, err = slots.getPeers(protocol).getRandom(1, nil)
-	require.Equal(t, err, ErrNoPeersAvailable)
+	require.Equal(t, err, utils.ErrNoPeersAvailable)
 	_, err = slots.getPeers(protocol1).getRandom(1, nil)
-	require.Equal(t, err, ErrNoPeersAvailable)
+	require.Equal(t, err, utils.ErrNoPeersAvailable)
 }

waku/v2/protocol/enr/enr.go

@@ -28,6 +28,23 @@ const ShardingBitVectorEnrField = "rsv"
 // WakuEnrBitfield is a8-bit flag field to indicate Waku capabilities. Only the 4 LSBs are currently defined according to RFC31 (https://rfc.vac.dev/spec/31/).
 type WakuEnrBitfield = uint8
 
+func GetWakuEnrBitField(node *enode.Node) (WakuEnrBitfield, error) {
+	enrField := []byte{}
+	err := node.Record().Load(enr.WithEntry(WakuENRField, &enrField))
+	if err != nil {
+		if enr.IsNotFound(err) {
+			return 0, nil
+		}
+		return 0, err
+	}
+
+	if len(enrField) == 0 {
+		return 0, err
+	}
+
+	return WakuEnrBitfield(enrField[0]), nil
+}
+
 // NewWakuEnrBitfield creates a WakuEnrBitField whose value will depend on which protocols are enabled in the node
 func NewWakuEnrBitfield(lightpush, filter, store, relay bool) WakuEnrBitfield {
 	var v uint8

waku/v2/utils/peer.go

@@ -1,6 +1,8 @@
 package utils
 
 import (
+	"errors"
+
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/multiformats/go-multiaddr"
 )
@@ -10,6 +12,10 @@ type DialError struct {
 	PeerID peer.ID
 }
 
+// ErrNoPeersAvailable is emitted when no suitable peers are found for
+// some protocol
+var ErrNoPeersAvailable = errors.New("no suitable peers found")
+
 // GetPeerID is used to extract the peerID from a multiaddress
 func GetPeerID(m multiaddr.Multiaddr) (peer.ID, error) {
 	peerIDStr, err := m.ValueForProtocol(multiaddr.P_P2P)