chore(rln): run rln in all relay pubsubtopics + remove cli flags

cmd/waku/flags_rln.go

@@ -6,7 +6,6 @@ package main
 import (
 	cli "github.com/urfave/cli/v2"
 	wcli "github.com/waku-org/go-waku/waku/cliutils"
-	"github.com/waku-org/go-waku/waku/v2/protocol"
 )
 
 func rlnFlags() []cli.Flag {
@@ -23,18 +22,6 @@ func rlnFlags() []cli.Flag {
 			Usage:       "the index of credentials to use, within a specific rln membership set",
 			Destination: &options.RLNRelay.MembershipGroupIndex,
 		},
-		&cli.StringFlag{
-			Name:        "rln-relay-pubsub-topic",
-			Value:       "/waku/2/default-waku/proto",
-			Usage:       "the pubsub topic for which rln-relay gets enabled",
-			Destination: &options.RLNRelay.PubsubTopic,
-		},
-		&cli.StringFlag{
-			Name:        "rln-relay-content-topic",
-			Value:       protocol.NewContentTopic("toy-chat", 3, "mingde", "proto").String(),
-			Usage:       "the content topic for which rln-relay gets enabled",
-			Destination: &options.RLNRelay.ContentTopic,
-		},
 		&cli.BoolFlag{
 			Name:        "rln-relay-dynamic",
 			Usage:       "Enable waku-rln-relay with on-chain dynamic group management",

cmd/waku/node_rln.go

@@ -17,13 +17,11 @@ func checkForRLN(logger *zap.Logger, options NodeOptions, nodeOpts *[]node.WakuN
 			failOnErr(errors.New("relay not available"), "Could not enable RLN Relay")
 		}
 		if !options.RLNRelay.Dynamic {
-			*nodeOpts = append(*nodeOpts, node.WithStaticRLNRelay(options.RLNRelay.PubsubTopic, options.RLNRelay.ContentTopic, rln.MembershipIndex(options.RLNRelay.MembershipGroupIndex), nil))
+			*nodeOpts = append(*nodeOpts, node.WithStaticRLNRelay(rln.MembershipIndex(options.RLNRelay.MembershipGroupIndex), nil))
 		} else {
 			// TODO: too many parameters in this function
 			// consider passing a config struct instead
 			*nodeOpts = append(*nodeOpts, node.WithDynamicRLNRelay(
-				options.RLNRelay.PubsubTopic,
-				options.RLNRelay.ContentTopic,
 				options.RLNRelay.CredentialsPath,
 				options.RLNRelay.CredentialsPassword,
 				options.RLNRelay.CredentialsIndex,

cmd/waku/options.go

@@ -40,8 +40,6 @@ type RLNRelayOptions struct {
 	CredentialsIndex          uint
 	TreePath                  string
 	MembershipGroupIndex      uint
-	PubsubTopic               string
-	ContentTopic              string
 	Dynamic                   bool
 	ETHClientAddress          string
 	MembershipContractAddress common.Address

examples/chat2/chat.go

@@ -141,7 +141,7 @@ func (c *Chat) receiveMessages() {
 		case value := <-c.C:
 
 			msgContentTopic := value.Message().ContentTopic
-			if msgContentTopic != c.options.ContentTopic || (c.options.RLNRelay.Enable && msgContentTopic != c.options.RLNRelay.ContentTopic) {
+			if msgContentTopic != c.options.ContentTopic {
 				continue // Discard messages from other topics
 			}
 

examples/chat2/exec.go

@@ -50,8 +50,6 @@ func execute(options Options) {
 		if options.RLNRelay.Dynamic {
 			fmt.Println("Setting up dynamic rln...")
 			opts = append(opts, node.WithDynamicRLNRelay(
-				options.RLNRelay.PubsubTopic,
-				options.RLNRelay.ContentTopic,
 				options.RLNRelay.CredentialsPath,
 				options.RLNRelay.CredentialsPassword,
 				options.RLNRelay.CredentialsIndex,
@@ -63,8 +61,6 @@ func execute(options Options) {
 			))
 		} else {
 			opts = append(opts, node.WithStaticRLNRelay(
-				options.RLNRelay.PubsubTopic,
-				options.RLNRelay.ContentTopic,
 				uint(options.RLNRelay.MembershipIndex),
 				spamHandler))
 		}

examples/chat2/flags.go

@@ -6,7 +6,6 @@ import (
 	"github.com/waku-org/go-waku/waku/cliutils"
 	wcli "github.com/waku-org/go-waku/waku/cliutils"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
-	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 
 	"github.com/urfave/cli/v2"
 )
@@ -198,18 +197,6 @@ func getFlags() []cli.Flag {
 			Usage:       "the index of credentials to use",
 			Destination: &options.RLNRelay.CredentialsIndex,
 		},
-		&cli.StringFlag{
-			Name:        "rln-relay-pubsub-topic",
-			Value:       relay.DefaultWakuTopic,
-			Usage:       "the pubsub topic for which rln-relay gets enabled",
-			Destination: &options.RLNRelay.PubsubTopic,
-		},
-		&cli.StringFlag{
-			Name:        "rln-relay-content-topic",
-			Value:       testnetContentTopic,
-			Usage:       "the content topic for which rln-relay gets enabled",
-			Destination: &options.RLNRelay.ContentTopic,
-		},
 		&cli.BoolFlag{
 			Name:        "rln-relay-dynamic",
 			Usage:       "Enable waku-rln-relay with on-chain dynamic group management",

examples/chat2/options.go

@@ -34,8 +34,6 @@ type RLNRelayOptions struct {
 	CredentialsIndex          uint
 	MembershipGroupIndex      uint
 	MembershipIndex           uint
-	PubsubTopic               string
-	ContentTopic              string
 	Dynamic                   bool
 	ETHClientAddress          string
 	MembershipContractAddress common.Address

waku/v2/node/wakunode2.go

@@ -10,6 +10,7 @@ import (
 	backoffv4 "github.com/cenkalti/backoff/v4"
 	golog "github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"go.uber.org/zap"
 
 	"github.com/ethereum/go-ethereum/crypto"
@@ -66,10 +67,14 @@ type IdentityCredential = struct {
 	IDCommitment byte32 `json:"idCommitment"`
 }
 
+type SpamHandler = func(message *pb.WakuMessage) error
+
 type RLNRelay interface {
 	IdentityCredential() (IdentityCredential, error)
 	MembershipIndex() (uint, error)
 	AppendRLNProof(msg *pb.WakuMessage, senderEpochTime time.Time) error
+	Validator(spamHandler SpamHandler) func(ctx context.Context, peerID peer.ID, message *pubsub.Message) bool
+	Start(ctx context.Context) error
 	Stop() error
 }
 
@@ -275,6 +280,14 @@ func New(opts ...WakuNodeOption) (*WakuNode, error) {
 	}
 
 	w.rendezvous = rendezvous.NewRendezvous(w.opts.rendezvousDB, w.peerConnector, w.log)
+
+	if w.opts.enableRelay {
+		err = w.setupRLNRelay()
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	w.relay = relay.NewWakuRelay(w.bcaster, w.opts.minRelayPeersToPublish, w.timesource, w.opts.prometheusReg, w.log, w.opts.pubsubOpts...)
 	w.legacyFilter = legacy_filter.NewWakuFilter(w.bcaster, w.opts.isLegacyFilterFullNode, w.timesource, w.opts.prometheusReg, w.log, w.opts.legacyFilterOpts...)
 	w.filterFullNode = filter.NewWakuFilterFullNode(w.timesource, w.opts.prometheusReg, w.log, w.opts.filterOpts...)
@@ -400,6 +413,13 @@ func (w *WakuNode) Start(ctx context.Context) error {
 		}
 	}
 
+	if w.opts.enableRLN {
+		err = w.startRlnRelay(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
 	w.relay.SetHost(host)
 
 	if w.opts.enableRelay {
@@ -479,13 +499,6 @@ func (w *WakuNode) Start(ctx context.Context) error {
 		}
 	}
 
-	if w.opts.enableRLN {
-		err = w.mountRlnRelay(ctx)
-		if err != nil {
-			return err
-		}
-	}
-
 	return nil
 }
 

waku/v2/node/wakunode2_no_rln.go

@@ -9,7 +9,11 @@ func (w *WakuNode) RLNRelay() RLNRelay {
 	return nil
 }
 
-func (w *WakuNode) mountRlnRelay(ctx context.Context) error {
+func (w *WakuNode) setupRLNRelay() error {
+	return nil
+}
+
+func (w *WakuNode) startRlnRelay(ctx context.Context) error {
 	return nil
 }
 

waku/v2/node/wakunode2_rln.go

@@ -8,28 +8,26 @@ import (
 	"context"
 	"errors"
 
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/waku-org/go-waku/waku/v2/protocol/rln"
 	"github.com/waku-org/go-waku/waku/v2/protocol/rln/group_manager/dynamic"
 	"github.com/waku-org/go-waku/waku/v2/protocol/rln/group_manager/static"
 	r "github.com/waku-org/go-zerokit-rln/rln"
-	"go.uber.org/zap"
 )
 
 // RLNRelay is used to access any operation related to Waku RLN protocol
 func (w *WakuNode) RLNRelay() RLNRelay {
 	return w.rlnRelay
 }
 
-func (w *WakuNode) mountRlnRelay(ctx context.Context) error {
-	// check whether inputs are provided
-	// relay protocol is the prerequisite of rln-relay
-	if w.Relay() == nil {
-		return errors.New("relay protocol is required")
-	}
-
+func (w *WakuNode) setupRLNRelay() error {
 	var err error
 	var groupManager rln.GroupManager
 
+	if !w.opts.enableRLN {
+		return nil
+	}
+
 	if !w.opts.rlnRelayDynamic {
 		w.log.Info("setting up waku-rln-relay in off-chain mode")
 
@@ -61,18 +59,27 @@ func (w *WakuNode) mountRlnRelay(ctx context.Context) error {
 		}
 	}
 
-	rlnRelay, err := rln.New(w.Relay(), groupManager, w.opts.rlnTreePath, w.opts.rlnRelayPubsubTopic, w.opts.rlnRelayContentTopic, w.opts.rlnSpamHandler, w.timesource, w.log)
+	rlnRelay, err := rln.New(groupManager, w.opts.rlnTreePath, w.timesource, w.log)
 	if err != nil {
 		return err
 	}
 
-	err = rlnRelay.Start(ctx)
+	w.rlnRelay = rlnRelay
+
+	// Adding RLN as a default validator
+	w.opts.pubsubOpts = append(w.opts.pubsubOpts, pubsub.WithDefaultValidator(rlnRelay.Validator(w.opts.rlnSpamHandler)))
+
+	return nil
+}
+
+func (w *WakuNode) startRlnRelay(ctx context.Context) error {
+	rlnRelay := w.rlnRelay.(*rln.WakuRLNRelay)
+
+	err := rlnRelay.Start(ctx)
 	if err != nil {
 		return err
 	}
 
-	w.rlnRelay = rlnRelay
-
 	if !w.opts.rlnRelayDynamic {
 		// check the correct construction of the tree by comparing the calculated root against the expected root
 		// no error should happen as it is already captured in the unit tests
@@ -91,7 +98,7 @@ func (w *WakuNode) mountRlnRelay(ctx context.Context) error {
 		}
 	}
 
-	w.log.Info("mounted waku RLN relay", zap.String("pubsubTopic", w.opts.rlnRelayPubsubTopic), zap.String("contentTopic", w.opts.rlnRelayContentTopic))
+	w.log.Info("mounted waku RLN relay")
 
 	return nil
 }

waku/v2/node/wakuoptions.go

@@ -95,8 +95,6 @@ type WakuNodeParameters struct {
 
 	enableRLN                    bool
 	rlnRelayMemIndex             uint
-	rlnRelayPubsubTopic          string
-	rlnRelayContentTopic         string
 	rlnRelayDynamic              bool
 	rlnSpamHandler               func(message *pb.WakuMessage) error
 	rlnETHClientAddress          string

waku/v2/node/wakuoptions_rln.go

@@ -11,29 +11,25 @@ import (
 
 // WithStaticRLNRelay enables the Waku V2 RLN protocol in offchain mode
 // Requires the `gowaku_rln` build constrain (or the env variable RLN=true if building go-waku)
-func WithStaticRLNRelay(pubsubTopic string, contentTopic string, memberIndex r.MembershipIndex, spamHandler rln.SpamHandler) WakuNodeOption {
+func WithStaticRLNRelay(memberIndex r.MembershipIndex, spamHandler rln.SpamHandler) WakuNodeOption {
 	return func(params *WakuNodeParameters) error {
 		params.enableRLN = true
 		params.rlnRelayDynamic = false
 		params.rlnRelayMemIndex = memberIndex
-		params.rlnRelayPubsubTopic = pubsubTopic
-		params.rlnRelayContentTopic = contentTopic
 		params.rlnSpamHandler = spamHandler
 		return nil
 	}
 }
 
 // WithDynamicRLNRelay enables the Waku V2 RLN protocol in onchain mode.
 // Requires the `gowaku_rln` build constrain (or the env variable RLN=true if building go-waku)
-func WithDynamicRLNRelay(pubsubTopic string, contentTopic string, keystorePath string, keystorePassword string, keystoreIndex uint, treePath string, membershipContract common.Address, membershipGroupIndex uint, spamHandler rln.SpamHandler, ethClientAddress string) WakuNodeOption {
+func WithDynamicRLNRelay(keystorePath string, keystorePassword string, keystoreIndex uint, treePath string, membershipContract common.Address, membershipGroupIndex uint, spamHandler rln.SpamHandler, ethClientAddress string) WakuNodeOption {
 	return func(params *WakuNodeParameters) error {
 		params.enableRLN = true
 		params.rlnRelayDynamic = true
 		params.keystorePassword = keystorePassword
 		params.keystorePath = keystorePath
 		params.keystoreIndex = keystoreIndex
-		params.rlnRelayPubsubTopic = pubsubTopic
-		params.rlnRelayContentTopic = contentTopic
 		params.rlnSpamHandler = spamHandler
 		params.rlnETHClientAddress = ethClientAddress
 		params.rlnMembershipContractAddress = membershipContract

waku/v2/protocol/relay/waku_relay.go

@@ -179,6 +179,7 @@ func NewWakuRelay(bcaster Broadcaster, minPeersToPublish int, timesource timesou
 		pubsub.WithSeenMessagesTTL(2 * time.Minute),
 		pubsub.WithPeerScore(w.peerScoreParams, w.peerScoreThresholds),
 		pubsub.WithPeerScoreInspect(w.peerScoreInspector, 6*time.Second),
+		// TODO: to improve - setup default validator only if no default validator has been set.
 		pubsub.WithDefaultValidator(func(ctx context.Context, peerID peer.ID, message *pubsub.Message) bool {
 			msg := new(pb.WakuMessage)
 			err := proto.Unmarshal(message.Data, msg)

waku/v2/protocol/rln/group_manager/dynamic/web3.go

@@ -107,7 +107,7 @@ func (gm *DynamicGroupManager) watchNewEvents(ctx context.Context, rlnContract *
 	}
 }
 
-const maxBatchSize = uint64(5000000) // TODO: tune this
+const maxBatchSize = uint64(5000)
 const additiveFactorMultiplier = 0.10
 const multiplicativeDecreaseDivisor = 2