chore: disable filter logs to prevent spam (#1275 )

fix: filter network change handling (#1270 )
fix: update criteria context on missing msg verifier start (#1268 )
2025-02-18 11:16:16 +05:30 · 2025-01-03 15:47:27 +05:30 · 2024-12-24 14:28:53 +05:30 · 2024-12-24 11:27:48 +08:00 · 2024-12-16 19:18:16 +05:30 · 2024-12-12 18:46:22 +08:00
40 changed files with 555 additions and 102 deletions
@@ -23,6 +23,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/filter"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
 	"github.com/waku-org/go-waku/waku/v2/utils"
+	"golang.org/x/time/rate"
 )

 func createNode(t *testing.T, opts ...node.WakuNodeOption) *node.WakuNode {
@@ -37,8 +38,8 @@ func createNode(t *testing.T, opts ...node.WakuNodeOption) *node.WakuNode {

 // node2 connects to node1
 func twoFilterConnectedNodes(t *testing.T, pubSubTopics ...string) (*node.WakuNode, *node.WakuNode) {
-	node1 := createNode(t, node.WithWakuFilterFullNode())  // full node filter
-	node2 := createNode(t, node.WithWakuFilterLightNode()) // light node filter
+	node1 := createNode(t, node.WithWakuFilterFullNode(filter.WithFullNodeRateLimiter(rate.Inf, 0))) // full node filter
+	node2 := createNode(t, node.WithWakuFilterLightNode())                                           // light node filter

 	node2.Host().Peerstore().AddAddr(node1.Host().ID(), tests.GetHostAddress(node1.Host()), peerstore.PermanentAddrTTL)
 	err := node2.Host().Peerstore().AddProtocols(node1.Host().ID(), filter.FilterSubscribeID_v20beta1)
@@ -26,7 +26,7 @@ pkgs.buildGo121Module {
  '' else "";

  # FIXME: This needs to be manually changed when updating modules.
-  vendorHash = "sha256-TrKlv3UHhFl+1HviEYFTmOpF+UiVdL6h53IkJXBFsRo=";
+  vendorHash = "sha256-yQ3anfZ/PU0M0KHiXqA9Ri8zFkg1nTYIk43jmcdGZYU=";

  # Fix for 'nix run' trying to execute 'go-waku'.
  meta = { mainProgram = "waku"; };
@@ -62,6 +62,7 @@ require (
 	github.com/ipfs/go-log/v2 v2.5.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
 github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
 github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
@@ -62,6 +62,7 @@ require (
 	github.com/ipfs/go-log/v2 v2.5.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -337,6 +337,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
 github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
 github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
@@ -70,6 +70,7 @@ require (
 	github.com/ipfs/go-cid v0.4.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -349,6 +349,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
 github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
 github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
@@ -57,6 +57,7 @@ require (
 	github.com/ipfs/go-cid v0.4.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
 github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
 github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
@@ -60,6 +60,7 @@ require (
 	github.com/ipfs/go-cid v0.4.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
 github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
 github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
@@ -58,6 +58,7 @@ require (
 	github.com/ipfs/go-log/v2 v2.5.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
 github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
 github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
@@ -40,6 +40,7 @@ require (
 	github.com/dustin/go-humanize v1.0.1
 	github.com/go-chi/chi/v5 v5.0.0
 	github.com/jackc/pgx/v5 v5.4.1
+	github.com/jellydator/ttlcache/v3 v3.3.0
 	github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0
 	github.com/waku-org/go-noise v0.0.4
 	github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59
@@ -949,6 +949,8 @@ github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0
 github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
 github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
 github.com/jinzhu/now v1.1.1/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
@@ -26,6 +26,7 @@ import (
 // filterSubscriptions is the map of filter subscription IDs to subscriptions

 const filterSubBatchSize = 90
+const initNetworkConnType = 255

 type appFilterMap map[string]filterConfig

@@ -43,6 +44,7 @@ type FilterManager struct {
 	filterConfigs          appFilterMap // map of application filterID to {aggregatedFilterID, application ContentFilter}
 	waitingToSubQueue      chan filterConfig
 	envProcessor           EnevelopeProcessor
+	networkConnType        byte
 }

 type SubDetails struct {
@@ -61,7 +63,8 @@ type EnevelopeProcessor interface {
 	OnNewEnvelope(env *protocol.Envelope) error
 }

-func NewFilterManager(ctx context.Context, logger *zap.Logger, minPeersPerFilter int, envProcessor EnevelopeProcessor, node *filter.WakuFilterLightNode, opts ...SubscribeOptions) *FilterManager {
+func NewFilterManager(ctx context.Context, logger *zap.Logger, minPeersPerFilter int,
+	envProcessor EnevelopeProcessor, node *filter.WakuFilterLightNode, opts ...SubscribeOptions) *FilterManager {
 	// This fn is being mocked in test
 	mgr := new(FilterManager)
 	mgr.ctx = ctx
@@ -75,6 +78,7 @@ func NewFilterManager(ctx context.Context, logger *zap.Logger, minPeersPerFilter
 	mgr.incompleteFilterBatch = make(map[string]filterConfig)
 	mgr.filterConfigs = make(appFilterMap)
 	mgr.waitingToSubQueue = make(chan filterConfig, 100)
+	mgr.networkConnType = initNetworkConnType

 	//parsing the subscribe params only to read the batchInterval passed.
 	mgr.params = new(subscribeParameters)
@@ -107,14 +111,12 @@ func (mgr *FilterManager) startFilterSubLoop() {
 				mgr.incompleteFilterBatch = make(map[string]filterConfig)
 				mgr.Unlock()
 			}
-			subs := mgr.node.Subscriptions()
-			mgr.logger.Debug("filter stats", zap.Int("agg filters count", len(mgr.filterSubscriptions)), zap.Int("filter subs count", len(subs)))
 		}
 	}
 }

-// addFilter method checks if there are existing waiting filters for the pubsubTopic to be subscribed and adds the new filter to the same batch
-// once batchlimit is hit, all filters are subscribed to and new batch is created.
+// SubscribeFilter method checks if there are existing waiting filters for the pubsubTopic to be subscribed and adds the new filter to the same batch
+// once batch-limit is hit, all filters are subscribed to and new batch is created.
 // if node is not online, then batch is pushed to a queue to be picked up later for subscription and new batch is created

 func (mgr *FilterManager) SubscribeFilter(filterID string, cf protocol.ContentFilter) {
@@ -162,6 +164,7 @@ func (mgr *FilterManager) subscribeAndRunLoop(f filterConfig) {
 	defer utils.LogOnPanic()
 	ctx, cancel := context.WithCancel(mgr.ctx)
 	config := FilterConfig{MaxPeers: mgr.minPeersPerFilter}
+
 	sub, err := Subscribe(ctx, mgr.node, f.contentFilter, config, mgr.logger, mgr.params)
 	mgr.Lock()
 	mgr.filterSubscriptions[f.ID] = SubDetails{cancel, sub}
@@ -180,36 +183,102 @@ func (mgr *FilterManager) NetworkChange() {
 	mgr.node.PingPeers() // ping all peers to check if subscriptions are alive
 }

+func (mgr *FilterManager) checkAndProcessQueue(pubsubTopic string) {
+	if len(mgr.waitingToSubQueue) > 0 {
+		for af := range mgr.waitingToSubQueue {
+			// TODO: change the below logic once topic specific health is implemented for lightClients
+			if pubsubTopic == "" || pubsubTopic == af.contentFilter.PubsubTopic {
+				// check if any filter subs are pending and subscribe them
+				mgr.logger.Debug("subscribing from filter queue", zap.String("filter-id", af.ID), zap.Stringer("content-filter", af.contentFilter))
+				go mgr.subscribeAndRunLoop(af)
+			} else {
+				mgr.waitingToSubQueue <- af
+			}
+			if len(mgr.waitingToSubQueue) == 0 {
+				mgr.logger.Debug("no pending subscriptions")
+				break
+			}
+		}
+	}
+}
+
+func (mgr *FilterManager) closeAndWait(wg *sync.WaitGroup, asub *SubDetails) {
+	defer wg.Done()
+	asub.cancel()
+	for {
+		env, ok := <-asub.sub.DataCh
+		if !ok {
+			mgr.logger.Debug("unsubscribed filter", zap.Strings("content-topics", asub.sub.ContentFilter.ContentTopics.ToList()))
+			return
+		}
+		// process any in-flight envelopes
+		err := mgr.envProcessor.OnNewEnvelope(env)
+		if err != nil {
+			mgr.logger.Error("invoking onNewEnvelopes error", zap.Error(err))
+		}
+	}
+}
+
+func (mgr *FilterManager) resubscribeAllSubscriptions() {
+	filterSubsCount := len(mgr.filterSubscriptions)
+	mgr.Lock()
+	mgr.logger.Debug("unsubscribing all filter subscriptions", zap.Int("subs-count", filterSubsCount))
+	var wg sync.WaitGroup
+	wg.Add(len(mgr.filterSubscriptions))
+
+	for _, asub := range mgr.filterSubscriptions {
+		go mgr.closeAndWait(&wg, &asub)
+	}
+	mgr.filterSubscriptions = make(map[string]SubDetails)
+
+	mgr.Unlock()
+
+	wg.Wait() //Waiting till all unsubs are done to avoid race between sub and unsub
+
+	mgr.logger.Debug("unsubscribed all filter subscriptions", zap.Int("subs-count", filterSubsCount))
+
+	// locking to protect filterConfigs map, can't lock while calling subscribe as same lock is acquired inside subscribe
+	mgr.Lock()
+	localMap := make(appFilterMap)
+	for filterID, config := range mgr.filterConfigs {
+		localMap[filterID] = config
+	}
+	mgr.Unlock()
+
+	for filterID, config := range localMap {
+		mgr.SubscribeFilter(filterID, config.contentFilter)
+	}
+
+}
+
 // OnConnectionStatusChange to be triggered when connection status change is detected either from offline to online or vice-versa
 // Note that pubsubTopic specific change can be triggered by specifying pubsubTopic,
 // if pubsubTopic is empty it indicates complete connection status change such as node went offline or came back online.
-func (mgr *FilterManager) OnConnectionStatusChange(pubsubTopic string, newStatus bool) {
+func (mgr *FilterManager) OnConnectionStatusChange(pubsubTopic string, newStatus bool, connType byte) {
 	subs := mgr.node.Subscriptions()
 	mgr.logger.Debug("inside on connection status change", zap.Bool("new-status", newStatus),
 		zap.Int("agg filters count", len(mgr.filterSubscriptions)), zap.Int("filter subs count", len(subs)))
+	/*
+		Checking for initialization condition because when filterManager is initialized networkConnType is set to 255 and when first time node goes online
+		the network conn type will be set and will trigger resubscribe which is not desired.
+		Change in connType refers to scenario where the localnode's network has changed e.g: a mobile switching between wifi and cellular,
+		this in-turn means ip address of the localnode has changed.
+		this can cause issues in filter-push where it never recovers and hence resubscribing all filters
+	*/
+	if mgr.networkConnType != initNetworkConnType &&
+		mgr.networkConnType != connType { //
+		// resubscribe all existing filters
+		go mgr.resubscribeAllSubscriptions()
+	}
 	if newStatus && !mgr.onlineChecker.IsOnline() { // switched from offline to Online
+		mgr.onlineChecker.SetOnline(newStatus)
 		mgr.NetworkChange()
 		mgr.logger.Debug("switching from offline to online")
 		mgr.Lock()
-		if len(mgr.waitingToSubQueue) > 0 {
-			for af := range mgr.waitingToSubQueue {
-				// TODO: change the below logic once topic specific health is implemented for lightClients
-				if pubsubTopic == "" || pubsubTopic == af.contentFilter.PubsubTopic {
-					// check if any filter subs are pending and subscribe them
-					mgr.logger.Debug("subscribing from filter queue", zap.String("filter-id", af.ID), zap.Stringer("content-filter", af.contentFilter))
-					go mgr.subscribeAndRunLoop(af)
-				} else {
-					mgr.waitingToSubQueue <- af
-				}
-				if len(mgr.waitingToSubQueue) == 0 {
-					mgr.logger.Debug("no pending subscriptions")
-					break
-				}
-			}
-		}
+		mgr.checkAndProcessQueue(pubsubTopic)
 		mgr.Unlock()
 	}
-
+	mgr.networkConnType = connType
 	mgr.onlineChecker.SetOnline(newStatus)
 }

@@ -161,9 +161,9 @@ func (s *FilterApiTestSuite) TestFilterManager() {
 	// Mock peers going down
 	s.LightNodeHost.Peerstore().RemovePeer(s.FullNodeHost.ID())

-	fm.OnConnectionStatusChange("", false)
+	fm.OnConnectionStatusChange("", false, 0)
 	time.Sleep(2 * time.Second)
-	fm.OnConnectionStatusChange("", true)
+	fm.OnConnectionStatusChange("", true, 0)
 	s.ConnectToFullNode(s.LightNode, s.FullNode)
 	time.Sleep(3 * time.Second)

@@ -35,18 +35,21 @@ type MessageTracker interface {
 // MissingMessageVerifier is used to periodically retrieve missing messages from store nodes that have some specific criteria
 type MissingMessageVerifier struct {
 	ctx    context.Context
+	cancel context.CancelFunc
 	params missingMessageVerifierParams

 	storenodeRequestor common.StorenodeRequestor
 	messageTracker     MessageTracker

-	criteriaInterest   map[string]criteriaInterest // Track message verification requests and when was the last time a pubsub topic was verified for missing messages
+	criteriaInterest   map[string]*criteriaInterest // Track message verification requests and when was the last time a pubsub topic was verified for missing messages
 	criteriaInterestMu sync.RWMutex

-	C <-chan *protocol.Envelope
+	C chan *protocol.Envelope

-	timesource timesource.Timesource
-	logger     *zap.Logger
+	timesource   timesource.Timesource
+	logger       *zap.Logger
+	isRunning    bool
+	runningMutex sync.RWMutex
 }

 // NewMissingMessageVerifier creates an instance of a MissingMessageVerifier
@@ -63,6 +66,8 @@ func NewMissingMessageVerifier(storenodeRequester common.StorenodeRequestor, mes
 		messageTracker:     messageTracker,
 		logger:             logger.Named("missing-msg-verifier"),
 		params:             params,
+		criteriaInterest:   make(map[string]*criteriaInterest),
+		C:                  make(chan *protocol.Envelope, 1000),
 	}
 }

@@ -94,15 +99,36 @@ func (m *MissingMessageVerifier) SetCriteriaInterest(peerID peer.ID, contentFilt
 		currMessageVerificationRequest.cancel()
 	}

-	m.criteriaInterest[contentFilter.PubsubTopic] = criteriaInterest
+	m.criteriaInterest[contentFilter.PubsubTopic] = &criteriaInterest
+}
+
+func (m *MissingMessageVerifier) setRunning(running bool) {
+	m.runningMutex.Lock()
+	defer m.runningMutex.Unlock()
+	m.isRunning = running
 }

 func (m *MissingMessageVerifier) Start(ctx context.Context) {
-	m.ctx = ctx
-	m.criteriaInterest = make(map[string]criteriaInterest)
+	m.runningMutex.Lock()
+	if m.isRunning { //make sure verifier only runs once.
+		m.runningMutex.Unlock()
+		return
+	}
+	m.isRunning = true
+	m.runningMutex.Unlock()

-	c := make(chan *protocol.Envelope, 1000)
-	m.C = c
+	ctx, cancelFunc := context.WithCancel(ctx)
+	m.ctx = ctx
+	m.cancel = cancelFunc
+
+	// updating context for existing criteria
+	m.criteriaInterestMu.Lock()
+	for _, value := range m.criteriaInterest {
+		ctx, cancel := context.WithCancel(m.ctx)
+		value.ctx = ctx
+		value.cancel = cancel
+	}
+	m.criteriaInterestMu.Unlock()

 	go func() {
 		defer utils.LogOnPanic()
@@ -117,30 +143,39 @@ func (m *MissingMessageVerifier) Start(ctx context.Context) {
 				m.criteriaInterestMu.RLock()
 				critIntList := make([]criteriaInterest, 0, len(m.criteriaInterest))
 				for _, value := range m.criteriaInterest {
-					critIntList = append(critIntList, value)
+					critIntList = append(critIntList, *value)
 				}
 				m.criteriaInterestMu.RUnlock()
 				for _, interest := range critIntList {
 					select {
 					case <-ctx.Done():
+						m.setRunning(false)
 						return
 					default:
 						semaphore <- struct{}{}
 						go func(interest criteriaInterest) {
 							defer utils.LogOnPanic()
-							m.fetchHistory(c, interest)
+							m.fetchHistory(m.C, interest)
 							<-semaphore
 						}(interest)
 					}
 				}

 			case <-ctx.Done():
+				m.setRunning(false)
 				return
 			}
 		}
 	}()
 }

+func (m *MissingMessageVerifier) Stop() {
+	m.cancel()
+	m.runningMutex.Lock()
+	defer m.runningMutex.Unlock()
+	m.isRunning = false
+}
+
 func (m *MissingMessageVerifier) fetchHistory(c chan<- *protocol.Envelope, interest criteriaInterest) {
 	contentTopics := interest.contentFilter.ContentTopics.ToList()
 	for i := 0; i < len(contentTopics); i += maxContentTopicsPerRequest {
@@ -6,16 +6,15 @@ import (
 	"time"

 	"github.com/ethereum/go-ethereum/common"
+	"github.com/libp2p/go-libp2p/core/event"
+	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
 	"go.uber.org/zap"
-	"golang.org/x/time/rate"
 )

 const DefaultPeersToPublishForLightpush = 2
-const DefaultPublishingLimiterRate = rate.Limit(2)
-const DefaultPublishingLimitBurst = 4

 type PublishMethod int

@@ -51,8 +50,14 @@ type MessageSender struct {
 	publishMethod    PublishMethod
 	publisher        Publisher
 	messageSentCheck ISentCheck
-	rateLimiter      *PublishRateLimiter
+	rateLimiter      PublishRateLimiter
 	logger           *zap.Logger
+	evtMessageSent   event.Emitter
+}
+
+type MessageSent struct {
+	Size      uint32 // Size of payload in bytes
+	Timestamp int64
 }

 type Request struct {
@@ -74,14 +79,19 @@ func (r *Request) WithPublishMethod(publishMethod PublishMethod) *Request {
 	return r
 }

-func NewMessageSender(publishMethod PublishMethod, publisher Publisher, logger *zap.Logger) (*MessageSender, error) {
+func NewMessageSender(publishMethod PublishMethod, publisher Publisher, rateLimiter PublishRateLimiter, logger *zap.Logger) (*MessageSender, error) {
 	if publishMethod == UnknownMethod {
 		return nil, errors.New("publish method is required")
 	}
+
+	if rateLimiter == nil {
+		rateLimiter = NewDefaultRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst)
+	}
+
 	return &MessageSender{
 		publishMethod: publishMethod,
 		publisher:     publisher,
-		rateLimiter:   NewPublishRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst),
+		rateLimiter:   rateLimiter,
 		logger:        logger,
 	}, nil
 }
@@ -91,11 +101,20 @@ func (ms *MessageSender) WithMessageSentCheck(messageSentCheck ISentCheck) *Mess
 	return ms
 }

-func (ms *MessageSender) WithRateLimiting(rateLimiter *PublishRateLimiter) *MessageSender {
+func (ms *MessageSender) WithRateLimiting(rateLimiter PublishRateLimiter) *MessageSender {
 	ms.rateLimiter = rateLimiter
 	return ms
 }

+func (ms *MessageSender) WithMessageSentEmitter(host host.Host) *MessageSender {
+	evtMessageSent, err := host.EventBus().Emitter(new(MessageSent))
+	if err != nil {
+		ms.logger.Error("failed to create message sent emitter", zap.Error(err))
+	}
+	ms.evtMessageSent = evtMessageSent
+	return ms
+}
+
 func (ms *MessageSender) Send(req *Request) error {
 	logger := ms.logger.With(
 		zap.Stringer("envelopeHash", req.envelope.Hash()),
@@ -149,6 +168,16 @@ func (ms *MessageSender) Send(req *Request) error {
 		)
 	}

+	if ms.evtMessageSent != nil {
+		err := ms.evtMessageSent.Emit(MessageSent{
+			Size:      uint32(len(req.envelope.Message().Payload)),
+			Timestamp: req.envelope.Message().GetTimestamp(),
+		})
+		if err != nil {
+			logger.Error("failed to emit message sent event", zap.Error(err))
+		}
+	}
+
 	return nil
 }

@@ -3,6 +3,7 @@ package publish
 import (
 	"context"
 	"crypto/rand"
+	"sync"
 	"testing"
 	"time"

@@ -40,7 +41,7 @@ func (m *MockMessageSentCheck) Start() {
 }

 func TestNewSenderWithUnknownMethod(t *testing.T) {
-	sender, err := NewMessageSender(UnknownMethod, nil, nil)
+	sender, err := NewMessageSender(UnknownMethod, nil, nil, nil)
 	require.NotNil(t, err)
 	require.Nil(t, sender)
 }
@@ -54,7 +55,7 @@ func TestNewSenderWithRelay(t *testing.T) {
 	_, err = relayNode.Subscribe(context.Background(), protocol.NewContentFilter("test-pubsub-topic"))
 	require.Nil(t, err)
 	publisher := NewDefaultPublisher(nil, relayNode)
-	sender, err := NewMessageSender(Relay, publisher, utils.Logger())
+	sender, err := NewMessageSender(Relay, publisher, nil, utils.Logger())
 	require.Nil(t, err)
 	require.NotNil(t, sender)
 	require.Nil(t, sender.messageSentCheck)
@@ -80,7 +81,7 @@ func TestNewSenderWithRelayAndMessageSentCheck(t *testing.T) {
 	_, err = relayNode.Subscribe(context.Background(), protocol.NewContentFilter("test-pubsub-topic"))
 	require.Nil(t, err)
 	publisher := NewDefaultPublisher(nil, relayNode)
-	sender, err := NewMessageSender(Relay, publisher, utils.Logger())
+	sender, err := NewMessageSender(Relay, publisher, nil, utils.Logger())

 	check := &MockMessageSentCheck{Messages: make(map[string]map[common.Hash]uint32)}
 	sender.WithMessageSentCheck(check)
@@ -110,7 +111,7 @@ func TestNewSenderWithRelayAndMessageSentCheck(t *testing.T) {
 }

 func TestNewSenderWithLightPush(t *testing.T) {
-	sender, err := NewMessageSender(LightPush, nil, nil)
+	sender, err := NewMessageSender(LightPush, nil, nil, nil)
 	require.Nil(t, err)
 	require.NotNil(t, sender)
 	require.Equal(t, LightPush, sender.publishMethod)
@@ -129,3 +130,46 @@ func createRelayNode(t *testing.T) (host.Host, *relay.WakuRelay) {

 	return host, relay
 }
+
+func TestMessageSentEmitter(t *testing.T) {
+	host, relayNode := createRelayNode(t)
+	err := relayNode.Start(context.Background())
+	require.Nil(t, err)
+	defer relayNode.Stop()
+
+	_, err = relayNode.Subscribe(context.Background(), protocol.NewContentFilter("test-pubsub-topic"))
+	require.Nil(t, err)
+	publisher := NewDefaultPublisher(nil, relayNode)
+	sender, err := NewMessageSender(Relay, publisher, nil, utils.Logger())
+	require.Nil(t, err)
+
+	check := &MockMessageSentCheck{Messages: make(map[string]map[common.Hash]uint32)}
+	sender.WithMessageSentCheck(check)
+	sender.WithMessageSentEmitter(host)
+
+	msg := &pb.WakuMessage{
+		Payload:      []byte{1, 2, 3},
+		Timestamp:    utils.GetUnixEpoch(),
+		ContentTopic: "test-content-topic",
+	}
+	envelope := protocol.NewEnvelope(msg, *utils.GetUnixEpoch(), "test-pubsub-topic")
+	req := NewRequest(context.TODO(), envelope)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	sub, err := host.EventBus().Subscribe(new(MessageSent))
+	require.Nil(t, err)
+	defer sub.Close()
+
+	go func() {
+		for msgSentEvt := range sub.Out() {
+			msgSent := msgSentEvt.(MessageSent)
+			require.Equal(t, uint32(len(msg.Payload)), msgSent.Size)
+			wg.Done()
+		}
+	}()
+
+	err = sender.Send(req)
+	require.Nil(t, err)
+	go wg.Wait()
+}
@@ -9,22 +9,30 @@ import (
 	"golang.org/x/time/rate"
 )

+const DefaultPublishingLimiterRate = rate.Limit(5)
+const DefaultPublishingLimitBurst = 10
+
+// RateLimiter
+type PublishRateLimiter interface {
+	Check(ctx context.Context, logger *zap.Logger) error
+}
+
 // PublishRateLimiter is used to decorate publish functions to limit the
 // number of messages per second that can be published
-type PublishRateLimiter struct {
+type DefaultRateLimiter struct {
 	limiter *rate.Limiter
 }

 // NewPublishRateLimiter will create a new instance of PublishRateLimiter.
 // You can specify an rate.Inf value to in practice ignore the rate limiting
-func NewPublishRateLimiter(r rate.Limit, b int) *PublishRateLimiter {
-	return &PublishRateLimiter{
+func NewDefaultRateLimiter(r rate.Limit, b int) *DefaultRateLimiter {
+	return &DefaultRateLimiter{
 		limiter: rate.NewLimiter(r, b),
 	}
 }

 // ThrottlePublishFn is used to decorate a PublishFn so rate limiting is applied
-func (p *PublishRateLimiter) ThrottlePublishFn(ctx context.Context, publishFn PublishFn) PublishFn {
+func (p *DefaultRateLimiter) ThrottlePublishFn(ctx context.Context, publishFn PublishFn) PublishFn {
 	return func(envelope *protocol.Envelope, logger *zap.Logger) error {
 		if err := p.Check(ctx, logger); err != nil {
 			return err
@@ -33,7 +41,7 @@ func (p *PublishRateLimiter) ThrottlePublishFn(ctx context.Context, publishFn Pu
 	}
 }

-func (p *PublishRateLimiter) Check(ctx context.Context, logger *zap.Logger) error {
+func (p *DefaultRateLimiter) Check(ctx context.Context, logger *zap.Logger) error {
 	if err := p.limiter.Wait(ctx); err != nil {
 		if !errors.Is(err, context.Canceled) {
 			logger.Error("could not send message (limiter)", zap.Error(err))
@@ -14,7 +14,7 @@ import (
 )

 func TestRateLimit(t *testing.T) {
-	r := NewPublishRateLimiter(rate.Limit(1), 1)
+	r := NewDefaultRateLimiter(rate.Limit(1), 1)
 	l := utils.Logger()

 	var counter atomic.Int32
@@ -0,0 +1,63 @@
+package publish
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+var ErrRateLimited = errors.New("rate limit exceeded")
+
+const RlnLimiterCapacity = 100
+const RlnLimiterRefillInterval = 10 * time.Minute
+
+// RlnRateLimiter is used to rate limit the outgoing messages,
+// The capacity and refillInterval comes from RLN contract configuration.
+type RlnRateLimiter struct {
+	mu             sync.Mutex
+	capacity       int
+	tokens         int
+	refillInterval time.Duration
+	lastRefill     time.Time
+}
+
+// NewRlnPublishRateLimiter creates a new rate limiter, starts with a full capacity bucket.
+func NewRlnRateLimiter(capacity int, refillInterval time.Duration) *RlnRateLimiter {
+	return &RlnRateLimiter{
+		capacity:       capacity,
+		tokens:         capacity, // Start with a full bucket
+		refillInterval: refillInterval,
+		lastRefill:     time.Now(),
+	}
+}
+
+// Allow checks if a token can be consumed, and refills the bucket if necessary
+func (rl *RlnRateLimiter) Allow() bool {
+	rl.mu.Lock()
+	defer rl.mu.Unlock()
+
+	// Refill tokens if the refill interval has passed
+	now := time.Now()
+	if now.Sub(rl.lastRefill) >= rl.refillInterval {
+		rl.tokens = rl.capacity // Refill the bucket
+		rl.lastRefill = now
+	}
+
+	// Check if there are tokens available
+	if rl.tokens > 0 {
+		rl.tokens--
+		return true
+	}
+
+	return false
+}
+
+func (rl *RlnRateLimiter) Check(ctx context.Context, logger *zap.Logger) error {
+	if rl.Allow() {
+		return nil
+	}
+	return ErrRateLimited
+}
@@ -0,0 +1,26 @@
+package publish
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/waku-org/go-waku/waku/v2/utils"
+)
+
+func TestRlnRateLimit(t *testing.T) {
+	r := NewRlnRateLimiter(3, 5*time.Second)
+	l := utils.Logger()
+
+	for i := 0; i < 3; i++ {
+		require.NoError(t, r.Check(context.Background(), l))
+	}
+	require.ErrorIs(t, r.Check(context.Background(), l), ErrRateLimited)
+
+	time.Sleep(6 * time.Second)
+	for i := 0; i < 3; i++ {
+		require.NoError(t, r.Check(context.Background(), l))
+	}
+	require.ErrorIs(t, r.Check(context.Background(), l), ErrRateLimited)
+}
@@ -102,7 +102,6 @@ const maxFailedAttempts = 5
 const prunePeerStoreInterval = 10 * time.Minute
 const peerConnectivityLoopSecs = 15
 const maxConnsToPeerRatio = 3
-const badPeersCleanupInterval = 1 * time.Minute
 const maxDialFailures = 2

 // 80% relay peers 20% service peers
@@ -258,14 +257,13 @@ func (pm *PeerManager) Start(ctx context.Context) {
 	}
 }

-func (pm *PeerManager) removeBadPeers() {
-	if !pm.RelayEnabled {
-		for _, peerID := range pm.host.Peerstore().Peers() {
-			if pm.host.Peerstore().(wps.WakuPeerstore).ConnFailures(peerID) > maxDialFailures {
-				//delete peer from peerStore
-				pm.logger.Debug("removing bad peer due to recurring dial failures", zap.Stringer("peerID", peerID))
-				pm.RemovePeer(peerID)
-			}
+func (pm *PeerManager) CheckAndRemoveBadPeer(peerID peer.ID) {
+	if pm.host.Peerstore().(wps.WakuPeerstore).ConnFailures(peerID) > maxDialFailures &&
+		pm.peerConnector.onlineChecker.IsOnline() {
+		if origin, _ := pm.host.Peerstore().(wps.WakuPeerstore).Origin(peerID); origin != wps.Static { // delete only if a peer is discovered and not configured statically.
+			//delete peer from peerStore
+			pm.logger.Debug("removing bad peer due to recurring dial failures", zap.Stringer("peerID", peerID))
+			pm.RemovePeer(peerID)
 		}
 	}
 }
@@ -273,17 +271,13 @@ func (pm *PeerManager) removeBadPeers() {
 func (pm *PeerManager) peerStoreLoop(ctx context.Context) {
 	defer utils.LogOnPanic()
 	t := time.NewTicker(prunePeerStoreInterval)
-	t1 := time.NewTicker(badPeersCleanupInterval)
 	defer t.Stop()
-	defer t1.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-t.C:
 			pm.prunePeerStore()
-		case <-t1.C:
-			pm.removeBadPeers()
 		}
 	}
 }
@@ -749,6 +743,7 @@ func (pm *PeerManager) HandleDialError(err error, peerID peer.ID) {
 	if err == nil || errors.Is(err, context.Canceled) {
 		return
 	}
+
 	if pm.peerConnector != nil {
 		pm.peerConnector.addConnectionBackoff(peerID)
 	}
@@ -762,9 +757,4 @@ func (pm *PeerManager) HandleDialError(err error, peerID peer.ID) {
 			pm.logger.Error("failed to emit DialError", zap.Error(emitterErr))
 		}
 	}
-	if !pm.RelayEnabled && pm.host.Peerstore().(wps.WakuPeerstore).ConnFailures(peerID) >= maxDialFailures {
-		//delete peer from peerStore
-		pm.logger.Debug("removing bad peer due to recurring dial failures", zap.Stringer("peerID", peerID))
-		pm.RemovePeer(peerID)
-	}
 }
@@ -130,6 +130,7 @@ func (pm *PeerManager) selectServicePeer(criteria PeerSelectionCriteria) (PeerSe
 			if len(criteria.PubsubTopics) == 0 || (len(criteria.PubsubTopics) == 1 && criteria.PubsubTopics[0] == "") {
 				return slot.getRandom(criteria.MaxPeers, criteria.ExcludePeers)
 			} else { //PubsubTopic based selection
+				slot.mu.RLock()
 				keys := make([]peer.ID, 0, len(slot.m))
 				for i := range slot.m {
 					if PeerInSet(criteria.ExcludePeers, i) {
@@ -137,6 +138,7 @@ func (pm *PeerManager) selectServicePeer(criteria PeerSelectionCriteria) (PeerSe
 					}
 					keys = append(keys, i)
 				}
+				slot.mu.RUnlock()
 				selectedPeers := pm.host.Peerstore().(wps.WakuPeerstore).PeersByPubSubTopics(criteria.PubsubTopics, keys...)
 				tmpPeers, err := selectRandomPeers(selectedPeers, criteria.ExcludePeers, criteria.MaxPeers)
 				for tmpPeer := range tmpPeers {
@@ -15,6 +15,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	libp2pProtocol "github.com/libp2p/go-libp2p/core/protocol"
+	"github.com/libp2p/go-libp2p/p2p/net/swarm"
 	"github.com/libp2p/go-msgio/pbio"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/waku-org/go-waku/logging"
@@ -54,6 +55,7 @@ type WakuFilterLightNode struct {
 	log              *zap.Logger
 	subscriptions    *subscription.SubscriptionsMap
 	pm               *peermanager.PeerManager
+	limiter          *utils.RateLimiter
 	peerPingInterval time.Duration
 }

@@ -89,6 +91,7 @@ func NewWakuFilterLightNode(
 	onlineChecker onlinechecker.OnlineChecker,
 	reg prometheus.Registerer,
 	log *zap.Logger,
+	opts ...LightNodeOption,
 ) *WakuFilterLightNode {
 	wf := new(WakuFilterLightNode)
 	wf.log = log.Named("filterv2-lightnode")
@@ -99,6 +102,14 @@ func NewWakuFilterLightNode(
 	wf.CommonService = service.NewCommonService()
 	wf.metrics = newMetrics(reg)
 	wf.peerPingInterval = 1 * time.Minute
+
+	params := &LightNodeParameters{}
+	opts = append(DefaultLightNodeOptions(), opts...)
+	for _, opt := range opts {
+		opt(params)
+	}
+	wf.limiter = utils.NewRateLimiter(params.limitR, params.limitB)
+
 	return wf
 }

@@ -155,8 +166,16 @@ func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(network.Strea

 		logger := wf.log.With(logging.HostID("peerID", peerID))

+		if !wf.limiter.Allow(peerID) {
+			wf.metrics.RecordError(rateLimitFailure)
+			if err := stream.Reset(); err != nil {
+				wf.log.Error("resetting connection", zap.Error(err))
+			}
+			return
+		}
+
 		if !wf.subscriptions.IsSubscribedTo(peerID) {
-			logger.Warn("received message push from unknown peer", logging.HostID("peerID", peerID))
+			logger.Warn("received message push from unknown peer")
 			wf.metrics.RecordError(unknownPeerMessagePush)
 			//Send a wildcard unsubscribe to this peer so that further requests are not forwarded to us
 			if err := stream.Reset(); err != nil {
@@ -249,6 +268,10 @@ func (wf *WakuFilterLightNode) request(ctx context.Context, requestID []byte,
 		wf.metrics.RecordError(dialFailure)
 		if wf.pm != nil {
 			wf.pm.HandleDialError(err, peerID)
+			if errors.Is(err, swarm.ErrAllDialsFailed) ||
+				errors.Is(err, swarm.ErrDialBackoff) || errors.Is(err, swarm.ErrNoAddresses) {
+				wf.pm.CheckAndRemoveBadPeer(peerID)
+			}
 		}
 		return err
 	}
@@ -287,7 +310,7 @@ func (wf *WakuFilterLightNode) request(ctx context.Context, requestID []byte,

 	}

-	if filterSubscribeResponse.RequestId != request.RequestId {
+	if filterSubscribeResponse.RequestId != "N/A" && filterSubscribeResponse.RequestId != request.RequestId {
 		wf.log.Error("requestID mismatch", zap.String("expected", request.RequestId), zap.String("received", filterSubscribeResponse.RequestId))
 		wf.metrics.RecordError(requestIDMismatch)
 		err := NewFilterError(300, "request_id_mismatch")
@@ -337,7 +360,7 @@ func (wf *WakuFilterLightNode) handleFilterSubscribeOptions(ctx context.Context,
 	if params.pm != nil && reqPeerCount > 0 {

 		wf.log.Debug("handleFilterSubscribeOptions", zap.Int("peerCount", reqPeerCount), zap.Int("excludePeersLen", len(params.peersToExclude)))
-		params.selectedPeers, err = wf.pm.SelectPeers(
+		selectedPeers, err := wf.pm.SelectPeers(
 			peermanager.PeerSelectionCriteria{
 				SelectionType: params.peerSelectionType,
 				Proto:         FilterSubscribeID_v20beta1,
@@ -350,7 +373,12 @@ func (wf *WakuFilterLightNode) handleFilterSubscribeOptions(ctx context.Context,
 		)
 		if err != nil {
 			wf.log.Error("peer selection returned err", zap.Error(err))
-			return nil, nil, err
+			if len(params.selectedPeers) == 0 {
+				return nil, nil, err
+			}
+		}
+		if len(selectedPeers) > 0 {
+			params.selectedPeers = append(params.selectedPeers, selectedPeers...)
 		}
 	}
 	wf.log.Debug("handleFilterSubscribeOptions exit", zap.Int("selectedPeerCount", len(params.selectedPeers)))
@@ -24,8 +24,8 @@ func (wf *WakuFilterLightNode) PingPeer(peer peer.ID) {
 	ctxWithTimeout, cancel := context.WithTimeout(wf.CommonService.Context(), PingTimeout)
 	defer cancel()
 	err := wf.Ping(ctxWithTimeout, peer)
-	if err != nil {
-		wf.log.Warn("Filter ping failed towards peer", zap.Stringer("peer", peer), zap.Error(err))
+	if err != nil && wf.onlineChecker.IsOnline() {
+		wf.log.Info("Filter ping failed towards peer", zap.Stringer("peer", peer), zap.Error(err))
 		//quickly retry ping again before marking subscription as failure
 		//Note that PingTimeout is a fraction of PingInterval so this shouldn't cause parallel pings being sent.
 		ctxWithTimeout, cancel := context.WithTimeout(wf.CommonService.Context(), PingTimeout)
@@ -96,6 +96,7 @@ var (
 	peerNotFoundFailure        metricsErrCategory = "peer_not_found_failure"
 	writeResponseFailure       metricsErrCategory = "write_response_failure"
 	pushTimeoutFailure         metricsErrCategory = "push_timeout_failure"
+	rateLimitFailure           metricsErrCategory = "ratelimit_failure"
 )

 // RecordError increases the counter for different error types
@@ -11,6 +11,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/peermanager"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"go.uber.org/zap"
+	"golang.org/x/time/rate"
 )

 func (old *FilterSubscribeParameters) Copy() *FilterSubscribeParameters {
@@ -57,13 +58,35 @@ type (
 		Timeout        time.Duration
 		MaxSubscribers int
 		pm             *peermanager.PeerManager
+		limitR         rate.Limit
+		limitB         int
 	}

 	Option func(*FilterParameters)

+	LightNodeParameters struct {
+		limitR rate.Limit
+		limitB int
+	}
+
+	LightNodeOption func(*LightNodeParameters)
+
 	FilterSubscribeOption func(*FilterSubscribeParameters) error
 )

+func WithLightNodeRateLimiter(r rate.Limit, b int) LightNodeOption {
+	return func(params *LightNodeParameters) {
+		params.limitR = r
+		params.limitB = b
+	}
+}
+
+func DefaultLightNodeOptions() []LightNodeOption {
+	return []LightNodeOption{
+		WithLightNodeRateLimiter(15, 20),
+	}
+}
+
 func WithTimeout(timeout time.Duration) Option {
 	return func(params *FilterParameters) {
 		params.Timeout = timeout
@@ -202,9 +225,17 @@ func WithPeerManager(pm *peermanager.PeerManager) Option {
 	}
 }

+func WithFullNodeRateLimiter(r rate.Limit, b int) Option {
+	return func(params *FilterParameters) {
+		params.limitR = r
+		params.limitB = b
+	}
+}
+
 func DefaultOptions() []Option {
 	return []Option{
 		WithTimeout(DefaultIdleSubscriptionTimeout),
 		WithMaxSubscribers(DefaultMaxSubscribers),
+		WithFullNodeRateLimiter(1, 1),
 	}
 }
@@ -37,9 +37,9 @@ type (
 		metrics Metrics
 		log     *zap.Logger
 		*service.CommonService
-		subscriptions *SubscribersMap
-		pm            *peermanager.PeerManager
-
+		subscriptions    *SubscribersMap
+		pm               *peermanager.PeerManager
+		limiter          *utils.RateLimiter
 		maxSubscriptions int
 	}
 )
@@ -56,6 +56,7 @@ func NewWakuFilterFullNode(timesource timesource.Timesource, reg prometheus.Regi
 		opt(params)
 	}

+	wf.limiter = utils.NewRateLimiter(params.limitR, params.limitB)
 	wf.CommonService = service.NewCommonService()
 	wf.metrics = newMetrics(reg)
 	wf.subscriptions = NewSubscribersMap(params.Timeout)
@@ -93,7 +94,14 @@ func (wf *WakuFilterFullNode) start(sub *relay.Subscription) error {

 func (wf *WakuFilterFullNode) onRequest(ctx context.Context) func(network.Stream) {
 	return func(stream network.Stream) {
-		logger := wf.log.With(logging.HostID("peer", stream.Conn().RemotePeer()))
+		peerID := stream.Conn().RemotePeer()
+		logger := wf.log.With(logging.HostID("peer", peerID))
+
+		if !wf.limiter.Allow(peerID) {
+			wf.metrics.RecordError(rateLimitFailure)
+			wf.reply(ctx, stream, &pb.FilterSubscribeRequest{RequestId: "N/A"}, http.StatusTooManyRequests, "filter request rejected due rate limit exceeded")
+			return
+		}

 		reader := pbio.NewDelimitedReader(stream, math.MaxInt32)

@@ -22,6 +22,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/timesource"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
+	"golang.org/x/time/rate"
 )

 type LightNodeData struct {
@@ -133,7 +134,7 @@ func (s *FilterTestSuite) GetWakuFilterFullNode(topic string, withRegisterAll bo

 	nodeData := s.GetWakuRelay(topic)

-	node2Filter := NewWakuFilterFullNode(timesource.NewDefaultClock(), prometheus.DefaultRegisterer, s.Log)
+	node2Filter := NewWakuFilterFullNode(timesource.NewDefaultClock(), prometheus.DefaultRegisterer, s.Log, WithFullNodeRateLimiter(rate.Inf, 0))
 	node2Filter.SetHost(nodeData.FullNodeHost)

 	var sub *relay.Subscription
@@ -166,7 +167,7 @@ func (s *FilterTestSuite) GetWakuFilterLightNode() LightNodeData {
 	b := relay.NewBroadcaster(10)
 	s.Require().NoError(b.Start(context.Background()))
 	pm := peermanager.NewPeerManager(5, 5, nil, nil, true, s.Log)
-	filterPush := NewWakuFilterLightNode(b, pm, timesource.NewDefaultClock(), onlinechecker.NewDefaultOnlineChecker(true), prometheus.DefaultRegisterer, s.Log)
+	filterPush := NewWakuFilterLightNode(b, pm, timesource.NewDefaultClock(), onlinechecker.NewDefaultOnlineChecker(true), prometheus.DefaultRegisterer, s.Log, WithLightNodeRateLimiter(rate.Inf, 0))
 	filterPush.SetHost(host)
 	pm.SetHost(host)
 	return LightNodeData{filterPush, host}
@@ -13,6 +13,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	libp2pProtocol "github.com/libp2p/go-libp2p/core/protocol"
+	"github.com/libp2p/go-libp2p/p2p/net/swarm"
 	"github.com/libp2p/go-msgio/pbio"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/waku-org/go-waku/logging"
@@ -24,7 +25,6 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
-	"golang.org/x/time/rate"
 )

 // LightPushID_v20beta1 is the current Waku LightPush protocol identifier
@@ -40,7 +40,7 @@ var (
 type WakuLightPush struct {
 	h       host.Host
 	relay   *relay.WakuRelay
-	limiter *rate.Limiter
+	limiter *utils.RateLimiter
 	cancel  context.CancelFunc
 	pm      *peermanager.PeerManager
 	metrics Metrics
@@ -59,11 +59,12 @@ func NewWakuLightPush(relay *relay.WakuRelay, pm *peermanager.PeerManager, reg p
 	wakuLP.metrics = newMetrics(reg)

 	params := &LightpushParameters{}
+	opts = append(DefaultLightpushOptions(), opts...)
 	for _, opt := range opts {
 		opt(params)
 	}

-	wakuLP.limiter = params.limiter
+	wakuLP.limiter = utils.NewRateLimiter(params.limitR, params.limitB)

 	return wakuLP
 }
@@ -106,7 +107,7 @@ func (wakuLP *WakuLightPush) onRequest(ctx context.Context) func(network.Stream)
 			Response: &pb.PushResponse{},
 		}

-		if wakuLP.limiter != nil && !wakuLP.limiter.Allow() {
+		if !wakuLP.limiter.Allow(stream.Conn().RemotePeer()) {
 			wakuLP.metrics.RecordError(rateLimitFailure)
 			responseMsg := "exceeds the rate limit"
 			responsePushRPC.Response.Info = &responseMsg
@@ -198,6 +199,10 @@ func (wakuLP *WakuLightPush) request(ctx context.Context, req *pb.PushRequest, p
 		wakuLP.metrics.RecordError(dialFailure)
 		if wakuLP.pm != nil {
 			wakuLP.pm.HandleDialError(err, peerID)
+			if errors.Is(err, swarm.ErrAllDialsFailed) ||
+				errors.Is(err, swarm.ErrDialBackoff) || errors.Is(err, swarm.ErrNoAddresses) {
+				wakuLP.pm.CheckAndRemoveBadPeer(peerID)
+			}
 		}
 		return nil, err
 	}
@@ -14,7 +14,8 @@ import (
 )

 type LightpushParameters struct {
-	limiter *rate.Limiter
+	limitR rate.Limit
+	limitB int
 }

 type Option func(*LightpushParameters)
@@ -22,7 +23,14 @@ type Option func(*LightpushParameters)
 // WithRateLimiter is an option used to specify a rate limiter for requests received in lightpush protocol
 func WithRateLimiter(r rate.Limit, b int) Option {
 	return func(params *LightpushParameters) {
-		params.limiter = rate.NewLimiter(r, b)
+		params.limitR = r
+		params.limitB = b
+	}
+}
+
+func DefaultLightpushOptions() []Option {
+	return []Option{
+		WithRateLimiter(1, 1),
 	}
 }

@@ -8,6 +8,7 @@ import (
 	"time"

 	"github.com/waku-org/go-waku/waku/v2/peermanager"
+	"golang.org/x/time/rate"

 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peerstore"
@@ -273,7 +274,7 @@ func TestWakuLightPushCornerCases(t *testing.T) {
 	defer node2.Stop()
 	defer sub2.Unsubscribe()

-	lightPushNode2 := NewWakuLightPush(node2, pm, prometheus.DefaultRegisterer, utils.Logger())
+	lightPushNode2 := NewWakuLightPush(node2, pm, prometheus.DefaultRegisterer, utils.Logger(), WithRateLimiter(rate.Inf, 0))
 	lightPushNode2.SetHost(host2)
 	err := lightPushNode2.Start(ctx)
 	require.NoError(t, err)
@@ -358,7 +359,7 @@ func TestWakuLightPushWithStaticSharding(t *testing.T) {

 	clientHost, err := tests.MakeHost(context.Background(), port, rand.Reader)
 	require.NoError(t, err)
-	client := NewWakuLightPush(nil, nil, prometheus.DefaultRegisterer, utils.Logger())
+	client := NewWakuLightPush(nil, nil, prometheus.DefaultRegisterer, utils.Logger(), WithRateLimiter(rate.Inf, 0))
 	client.SetHost(clientHost)

 	// Node2
@@ -366,7 +367,7 @@ func TestWakuLightPushWithStaticSharding(t *testing.T) {
 	defer node2.Stop()
 	defer sub2.Unsubscribe()

-	lightPushNode2 := NewWakuLightPush(node2, nil, prometheus.DefaultRegisterer, utils.Logger())
+	lightPushNode2 := NewWakuLightPush(node2, nil, prometheus.DefaultRegisterer, utils.Logger(), WithRateLimiter(rate.Inf, 0))
 	lightPushNode2.SetHost(host2)
 	err = lightPushNode2.Start(ctx)
 	require.NoError(t, err)
@@ -23,7 +23,6 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/service"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
-	"golang.org/x/time/rate"
 )

 // PeerExchangeID_v20alpha1 is the current Waku Peer Exchange protocol identifier
@@ -51,7 +50,7 @@ type WakuPeerExchange struct {

 	peerConnector PeerConnector
 	enrCache      *enrCache
-	limiter       *rate.Limiter
+	limiter       *utils.RateLimiter
 }

 // NewWakuPeerExchange returns a new instance of WakuPeerExchange struct
@@ -68,11 +67,12 @@ func NewWakuPeerExchange(disc *discv5.DiscoveryV5, clusterID uint16, peerConnect
 	wakuPX.CommonService = service.NewCommonService()

 	params := &PeerExchangeParameters{}
+	opts = append(DefaultPeerExchangeOptions(), opts...)
 	for _, opt := range opts {
 		opt(params)
 	}

-	wakuPX.limiter = params.limiter
+	wakuPX.limiter = utils.NewRateLimiter(params.limiterR, params.limiterB)
 	return wakuPX, nil
 }

@@ -97,9 +97,10 @@ func (wakuPX *WakuPeerExchange) start() error {

 func (wakuPX *WakuPeerExchange) onRequest() func(network.Stream) {
 	return func(stream network.Stream) {
-		logger := wakuPX.log.With(logging.HostID("peer", stream.Conn().RemotePeer()))
+		peerID := stream.Conn().RemotePeer()
+		logger := wakuPX.log.With(logging.HostID("peer", peerID))

-		if wakuPX.limiter != nil && !wakuPX.limiter.Allow() {
+		if wakuPX.limiter != nil && !wakuPX.limiter.Allow(peerID) {
 			wakuPX.metrics.RecordError(rateLimitFailure)
 			wakuPX.log.Info("exceeds the rate limit")
 			// TODO: peer exchange protocol should contain an err field
@@ -12,7 +12,8 @@ import (
 )

 type PeerExchangeParameters struct {
-	limiter *rate.Limiter
+	limiterR rate.Limit
+	limiterB int
 }

 type Option func(*PeerExchangeParameters)
@@ -20,7 +21,14 @@ type Option func(*PeerExchangeParameters)
 // WithRateLimiter is an option used to specify a rate limiter for requests received in lightpush protocol
 func WithRateLimiter(r rate.Limit, b int) Option {
 	return func(params *PeerExchangeParameters) {
-		params.limiter = rate.NewLimiter(r, b)
+		params.limiterR = r
+		params.limiterB = b
+	}
+}
+
+func DefaultPeerExchangeOptions() []Option {
+	return []Option{
+		WithRateLimiter(1, 1),
 	}
 }

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"sync"

 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -73,6 +74,7 @@ type WakuStore struct {

 	defaultRatelimit rate.Limit
 	rateLimiters     map[peer.ID]*rate.Limiter
+	rateLimitersMux  sync.Mutex
 }

 // NewWakuStore is used to instantiate a StoreV3 client
@@ -297,11 +299,13 @@ func (s *WakuStore) queryFrom(ctx context.Context, storeRequest *pb.StoreQueryRe
 	logger.Debug("sending store request")

 	if !params.skipRatelimit {
+		s.rateLimitersMux.Lock()
 		rateLimiter, ok := s.rateLimiters[params.selectedPeer]
 		if !ok {
 			rateLimiter = rate.NewLimiter(s.defaultRatelimit, 1)
 			s.rateLimiters[params.selectedPeer] = rateLimiter
 		}
+		s.rateLimitersMux.Unlock()
 		err := rateLimiter.Wait(ctx)
 		if err != nil {
 			return nil, err
@@ -0,0 +1,69 @@
+package utils
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/jellydator/ttlcache/v3"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"golang.org/x/time/rate"
+)
+
+type RateLimiter struct {
+	sync.Mutex
+	limiters *ttlcache.Cache[peer.ID, *rate.Limiter]
+	r        rate.Limit
+	b        int
+}
+
+func NewRateLimiter(r rate.Limit, b int) *RateLimiter {
+	return &RateLimiter{
+		r: r,
+		b: b,
+		limiters: ttlcache.New[peer.ID, *rate.Limiter](
+			ttlcache.WithTTL[peer.ID, *rate.Limiter](30 * time.Minute),
+		),
+	}
+}
+
+func (r *RateLimiter) Start(ctx context.Context) {
+	go func() {
+		t := time.NewTicker(time.Hour)
+		defer t.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				r.Lock()
+				r.limiters.DeleteExpired()
+				r.Unlock()
+			}
+		}
+	}()
+}
+
+func (r *RateLimiter) getOrCreate(peerID peer.ID) *rate.Limiter {
+	r.Lock()
+	defer r.Unlock()
+
+	var limiter *rate.Limiter
+	if !r.limiters.Has(peerID) {
+		limiter = rate.NewLimiter(r.r, r.b)
+		r.limiters.Set(peerID, limiter, ttlcache.DefaultTTL)
+	} else {
+		v := r.limiters.Get(peerID)
+		limiter = v.Value()
+	}
+	return limiter
+}
+
+func (r *RateLimiter) Allow(peerID peer.ID) bool {
+	return r.getOrCreate(peerID).Allow()
+}
+
+func (r *RateLimiter) Wait(ctx context.Context, peerID peer.ID) error {
+	return r.getOrCreate(peerID).Wait(ctx)
+}
Author	SHA1	Message	Date
Prem Chaitanya Prathi	24932b529c	chore: disable filter logs to prevent spam (#1275 ) ci / env (push) Has been cancelled ci / lint (push) Has been cancelled ci / changes (push) Has been cancelled ci / build-macos-latest (push) Has been cancelled ci / build-ubuntu-latest (push) Has been cancelled ci / test-ci (push) Has been cancelled ci / test-with-race (push) Has been cancelled CodeQL / Analyze (go) (push) Has been cancelled	2025-02-18 11:16:16 +05:30
Prem Chaitanya Prathi	4ef460cb95	fix: filter network change handling (#1270 )	2025-01-03 15:47:27 +05:30
Prem Chaitanya Prathi	c0afa070a3	fix: update criteria context on missing msg verifier start (#1268 )	2024-12-24 14:28:53 +05:30
kaichao	6dcf177414	feat: rate limit with rln configuration (#1262 )	2024-12-24 11:27:48 +08:00
Prem Chaitanya Prathi	78b522db50	fix: have better defaults for send/recv rate-limits (#1267 )	2024-12-16 19:18:16 +05:30
frank	ffed0595ad	fix: concurrent peerMap iteration and write (#1266 )	2024-12-12 18:46:22 +08:00
Pablo Lopez	9a243696d7	fix: fatal error: concurrent map writes (#1265 )	2024-12-10 14:08:04 +02:00
Prem Chaitanya Prathi	809dba5854	fix: use bad peer removal logic only for lightpush and filter and option to restart missing message verifier (#1244 )	2024-12-09 14:14:28 +05:30
Arseniy Klempner	6550ff35bc	feat: option to emit event to bus upon message sent (#1261 )	2024-12-02 19:22:30 -08:00
richΛrd	0c594b3140	feat: filter rate limit (#1258 )	2024-11-29 12:25:55 -04:00