fix: and optimising fetching membership events (#706)

* fix: and optimising fetching membership events

* fix: start from lastProcessedBlock+1

* test: fetching membership logic

* refactor: usage of rlnInstance,rootTracker,groupManager

rlnInstance, rootTrack were previously created while creating rlnRelay
but were assigned to groupManager on Start of rlnRelay. This created
unncessary dependency of passing them to static and dynamic group
manager.
Web3Config uses interface EthClientI for client, so that we can pass
mock client for testing MembershipFetcher.

* fix: failing test

* fix: lint error

* fix: account for PR suggestions

* fix: failing race test

* fix: dont' increase fromBlock on error

* nit: fix naming and add comments

Author: harsh-98

waku/v2/node/wakunode2_rln.go

@@ -10,6 +10,7 @@ import (
 
 	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"
 	"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"
 	"github.com/waku-org/go-waku/waku/v2/protocol/rln/keystore"
@@ -23,12 +24,17 @@ func (w *WakuNode) RLNRelay() RLNRelay {
 
 func (w *WakuNode) setupRLNRelay() error {
 	var err error
-	var groupManager rln.GroupManager
 
 	if !w.opts.enableRLN {
 		return nil
 	}
 
+	var groupManager group_manager.GroupManager
+
+	rlnInstance, rootTracker, err := rln.GetRLNInstanceAndRootTracker(w.opts.rlnTreePath)
+	if err != nil {
+		return err
+	}
 	if !w.opts.rlnRelayDynamic {
 		w.log.Info("setting up waku-rln-relay in off-chain mode")
 
@@ -38,7 +44,8 @@ func (w *WakuNode) setupRLNRelay() error {
 			return err
 		}
 
-		groupManager, err = static.NewStaticGroupManager(groupKeys, idCredential, w.opts.rlnRelayMemIndex, w.log)
+		groupManager, err = static.NewStaticGroupManager(groupKeys, idCredential, w.opts.rlnRelayMemIndex, rlnInstance,
+			rootTracker, w.log)
 		if err != nil {
 			return err
 		}
@@ -57,17 +64,20 @@ func (w *WakuNode) setupRLNRelay() error {
 			appKeystore,
 			w.opts.keystorePassword,
 			w.opts.prometheusReg,
+			rlnInstance,
+			rootTracker,
 			w.log,
 		)
 		if err != nil {
 			return err
 		}
 	}
 
-	rlnRelay, err := rln.New(groupManager, w.opts.rlnTreePath, w.timesource, w.opts.prometheusReg, w.log)
-	if err != nil {
-		return err
-	}
+	rlnRelay := rln.New(group_manager.Details{
+		GroupManager: groupManager,
+		RootTracker:  rootTracker,
+		RLN:          rlnInstance,
+	}, w.timesource, w.opts.prometheusReg, w.log)
 
 	w.rlnRelay = rlnRelay
 

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

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"math/big"
-	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
@@ -28,28 +27,21 @@ var RLNAppInfo = keystore.AppInfo{
 }
 
 type DynamicGroupManager struct {
-	rln     *rln.RLN
-	log     *zap.Logger
+	MembershipFetcher
 	metrics Metrics
 
 	cancel context.CancelFunc
-	wg     sync.WaitGroup
 
 	identityCredential *rln.IdentityCredential
 	membershipIndex    rln.MembershipIndex
 
-	web3Config         *web3.Config
 	lastBlockProcessed uint64
 
-	eventHandler RegistrationEventHandler
-
 	appKeystore      *keystore.AppKeystore
 	keystorePassword string
-
-	rootTracker *group_manager.MerkleRootTracker
 }
 
-func handler(gm *DynamicGroupManager, events []*contracts.RLNMemberRegistered) error {
+func (gm *DynamicGroupManager) handler(events []*contracts.RLNMemberRegistered) error {
 	toRemoveTable := om.New()
 	toInsertTable := om.New()
 
@@ -116,18 +108,19 @@ func NewDynamicGroupManager(
 	appKeystore *keystore.AppKeystore,
 	keystorePassword string,
 	reg prometheus.Registerer,
+	rlnInstance *rln.RLN,
+	rootTracker *group_manager.MerkleRootTracker,
 	log *zap.Logger,
 ) (*DynamicGroupManager, error) {
 	log = log.Named("rln-dynamic")
 
+	web3Config := web3.NewConfig(ethClientAddr, memContractAddr)
 	return &DynamicGroupManager{
-		membershipIndex:  membershipIndex,
-		web3Config:       web3.NewConfig(ethClientAddr, memContractAddr),
-		eventHandler:     handler,
-		appKeystore:      appKeystore,
-		keystorePassword: keystorePassword,
-		log:              log,
-		metrics:          newMetrics(reg),
+		membershipIndex:   membershipIndex,
+		appKeystore:       appKeystore,
+		keystorePassword:  keystorePassword,
+		MembershipFetcher: NewMembershipFetcher(web3Config, rlnInstance, rootTracker, log),
+		metrics:           newMetrics(reg),
 	}, nil
 }
 
@@ -139,7 +132,7 @@ func (gm *DynamicGroupManager) memberExists(ctx context.Context, idCommitment rl
 	return gm.web3Config.RLNContract.MemberExists(&bind.CallOpts{Context: ctx}, rln.Bytes32ToBigInt(idCommitment))
 }
 
-func (gm *DynamicGroupManager) Start(ctx context.Context, rlnInstance *rln.RLN, rootTracker *group_manager.MerkleRootTracker) error {
+func (gm *DynamicGroupManager) Start(ctx context.Context) error {
 	if gm.cancel != nil {
 		return errors.New("already started")
 	}
@@ -154,9 +147,6 @@ func (gm *DynamicGroupManager) Start(ctx context.Context, rlnInstance *rln.RLN,
 		return err
 	}
 
-	gm.rln = rlnInstance
-	gm.rootTracker = rootTracker
-
 	// check if the contract exists by calling a static function
 	_, err = gm.getMembershipFee(ctx)
 	if err != nil {
@@ -168,7 +158,7 @@ func (gm *DynamicGroupManager) Start(ctx context.Context, rlnInstance *rln.RLN,
 		return err
 	}
 
-	if err = gm.HandleGroupUpdates(ctx, gm.eventHandler); err != nil {
+	if err = gm.MembershipFetcher.HandleGroupUpdates(ctx, gm.handler); err != nil {
 		return err
 	}
 
@@ -278,9 +268,7 @@ func (gm *DynamicGroupManager) Stop() error {
 		return err
 	}
 
-	gm.web3Config.ETHClient.Close()
-
-	gm.wg.Wait()
+	gm.MembershipFetcher.Stop()
 
 	return nil
 }

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

@@ -3,7 +3,6 @@ package dynamic
 import (
 	"context"
 	"math/big"
-	"sync"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/core/types"
@@ -35,21 +34,21 @@ func TestHandler(t *testing.T) {
 	rootTracker, err := group_manager.NewMerkleRootTracker(5, rlnInstance)
 	require.NoError(t, err)
 
-	ctx, cancel := context.WithCancel(context.TODO())
+	_, cancel := context.WithCancel(context.TODO())
 	defer cancel()
 
-	_ = ctx
-
 	gm := &DynamicGroupManager{
-		rln:    rlnInstance,
-		log:    utils.Logger(),
+		MembershipFetcher: NewMembershipFetcher(
+			&web3.Config{
+				ChainID: big.NewInt(1),
+			},
+			rlnInstance,
+			rootTracker,
+			utils.Logger(),
+		),
 		cancel: cancel,
-		wg:     sync.WaitGroup{},
-		web3Config: &web3.Config{
-			ChainID: big.NewInt(1),
-		},
-		rootTracker: rootTracker,
-		metrics:     newMetrics(prometheus.DefaultRegisterer),
+
+		metrics: newMetrics(prometheus.DefaultRegisterer),
 	}
 
 	root0 := [32]byte{62, 31, 25, 34, 223, 182, 113, 211, 249, 18, 247, 234, 70, 30, 10, 136, 238, 132, 143, 221, 225, 43, 108, 24, 171, 26, 210, 197, 106, 231, 52, 33}
@@ -59,7 +58,7 @@ func TestHandler(t *testing.T) {
 
 	events := []*contracts.RLNMemberRegistered{eventBuilder(1, false, 0xaaaa, 1)}
 
-	err = handler(gm, events)
+	err = gm.handler(events)
 	require.NoError(t, err)
 
 	roots = gm.rootTracker.Roots()
@@ -75,7 +74,7 @@ func TestHandler(t *testing.T) {
 		eventBuilder(4, false, 0xeeee, 5),
 	}
 
-	err = handler(gm, events)
+	err = gm.handler(events)
 	require.NoError(t, err)
 
 	// Root[1] should become [0]
@@ -99,7 +98,7 @@ func TestHandler(t *testing.T) {
 		eventBuilder(3, false, 0xeeee, 5),
 	}
 
-	err = handler(gm, events)
+	err = gm.handler(events)
 	require.NoError(t, err)
 
 	roots = gm.rootTracker.Roots()
@@ -113,7 +112,7 @@ func TestHandler(t *testing.T) {
 	// Adding multiple events for same block
 	events = []*contracts.RLNMemberRegistered{}
 
-	err = handler(gm, events)
+	err = gm.handler(events)
 	require.NoError(t, err)
 
 	roots = gm.rootTracker.Roots()