refactor: use peerInfo instead of peerID

fix: filter network change handling (#1270 )
fix: update criteria context on missing msg verifier start (#1268 )
2025-01-22 14:21:17 -04:00 · 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
62 changed files with 670 additions and 172 deletions
@@ -386,7 +386,7 @@ func Execute(options NodeOptions) error {
 	if options.PeerExchange.Enable && options.PeerExchange.Node != nil {
 		logger.Info("retrieving peer info via peer exchange protocol")

-		peerID, err := wakuNode.AddPeer(*options.PeerExchange.Node, wakupeerstore.Static,
+		peerID, err := wakuNode.AddPeer([]multiaddr.Multiaddr{*options.PeerExchange.Node}, wakupeerstore.Static,
 			pubSubTopicMapKeys, peer_exchange.PeerExchangeID_v20alpha1)
 		if err != nil {
 			logger.Error("adding peer exchange peer", logging.MultiAddrs("node", *options.PeerExchange.Node), zap.Error(err))
@@ -481,7 +481,7 @@ func processTopics(options NodeOptions) (map[string][]string, error) {

 func addStaticPeers(wakuNode *node.WakuNode, addresses []multiaddr.Multiaddr, pubSubTopics []string, protocols ...protocol.ID) error {
 	for _, addr := range addresses {
-		_, err := wakuNode.AddPeer(addr, wakupeerstore.Static, pubSubTopics, protocols...)
+		_, err := wakuNode.AddPeer([]multiaddr.Multiaddr{addr}, wakupeerstore.Static, pubSubTopics, protocols...)
 		if err != nil {
 			return fmt.Errorf("could not add static peer: %w", err)
 		}
@@ -7,6 +7,7 @@ import (
 	"github.com/go-chi/chi/v5"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
+	"github.com/multiformats/go-multiaddr"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/waku-org/go-waku/cmd/waku/server"
 	"github.com/waku-org/go-waku/logging"
@@ -117,7 +118,7 @@ func (a *AdminService) postV1Peer(w http.ResponseWriter, req *http.Request) {
 		protos = append(protos, protocol.ID(proto))
 	}

-	id, err := a.node.AddPeer(addr, peerstore.Static, topics, protos...)
+	id, err := a.node.AddPeer([]multiaddr.Multiaddr{addr}, peerstore.Static, topics, protos...)
 	if err != nil {
 		a.log.Error("failed to add peer", zap.Error(err))
 		writeErrOrResponse(w, err, nil)
@@ -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=
@@ -137,7 +137,7 @@ func Execute() error {
 	if err != nil {
 		log.Info("Error decoding multiaddr ", zap.Error(err))
 	}
-	peerID, err := lightNode.AddPeer(maddr, wps.Static,
+	peerID, err := lightNode.AddPeer([]multiaddr.Multiaddr{maddr}, wps.Static,
 		[]string{pubsubTopicStr}, filter.FilterSubscribeID_v20beta1, lightpush.LightPushID_v20beta1)
 	if err != nil {
 		log.Info("Error adding filter peer on light node ", zap.Error(err))
@@ -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=
@@ -138,7 +138,7 @@ func Execute() error {
 		if err != nil {
 			log.Info("Error decoding multiaddr ", zap.Error(err))
 		}
-		_, err = wakuNode.AddPeer(maddr, wps.Static,
+		_, err = wakuNode.AddPeer([]multiaddr.Multiaddr{maddr}, wps.Static,
 			[]string{pubsubTopicStr}, relay.WakuRelayID_v200)
 		if err != nil {
 			log.Info("Error adding filter peer on light node ", zap.Error(err))
@@ -117,6 +117,6 @@ func addPeer(wakuNode *node.WakuNode, addr *multiaddr.Multiaddr, topics []string
 	if addr == nil {
 		return nil
 	}
-	_, err := wakuNode.AddPeer(*addr, peerstore.Static, topics, protocols...)
+	_, err := wakuNode.AddPeer([]multiaddr.Multiaddr{*addr}, peerstore.Static, topics, protocols...)
 	return err
 }
@@ -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=
@@ -94,7 +94,7 @@ func main() {
 	// Setup filter
 	//

-	_, err = lightNode.AddPeer(fullNode.ListenAddresses()[0], wps.Static,
+	_, err = lightNode.AddPeer(fullNode.ListenAddresses(), wps.Static,
 		[]string{pubSubTopic.String()}, filter.FilterSubscribeID_v20beta1)
 	if err != nil {
 		log.Info("Error adding filter peer on light node ", err)
@@ -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=
@@ -352,7 +352,7 @@ func AddPeer(instance *WakuInstance, address string, protocolID string) (string,
 		return "", err
 	}

-	peerID, err := instance.node.AddPeer(ma, peerstore.Static, instance.relayTopics, libp2pProtocol.ID(protocolID))
+	peerID, err := instance.node.AddPeer([]multiaddr.Multiaddr{ma}, peerstore.Static, instance.relayTopics, libp2pProtocol.ID(protocolID))
 	if err != nil {
 		return "", err
 	}
@@ -8,5 +8,5 @@ import (
 )

 type StorenodeRequestor interface {
-	Query(ctx context.Context, peerID peer.ID, query *pb.StoreQueryRequest) (StoreRequestResult, error)
+	Query(ctx context.Context, peerInfo peer.AddrInfo, query *pb.StoreQueryRequest) (StoreRequestResult, error)
 }
@@ -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)
@@ -113,8 +117,8 @@ func (mgr *FilterManager) startFilterSubLoop() {
 	}
 }

-// 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 +166,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 +185,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)

@@ -98,6 +98,12 @@ func (m *StorenodeCycle) DisconnectActiveStorenode(backoff time.Duration) {
 func (m *StorenodeCycle) connectToNewStorenodeAndWait(ctx context.Context) error {
 	// Handle pinned storenodes
 	m.logger.Info("disconnecting storenode")
+
+	if m.storenodeConfigProvider == nil {
+		m.logger.Debug("storenodeConfigProvider not yet setup")
+		return nil
+	}
+
 	pinnedStorenode, err := m.storenodeConfigProvider.GetPinnedStorenode()
 	if err != nil {
 		m.logger.Error("could not obtain the pinned storenode", zap.Error(err))
@@ -252,6 +258,11 @@ func (m *StorenodeCycle) findNewStorenode(ctx context.Context) error {
 		}
 	}

+	if m.storenodeConfigProvider == nil {
+		m.logger.Debug("storenodeConfigProvider not yet setup")
+		return nil
+	}
+
 	pinnedStorenode, err := m.storenodeConfigProvider.GetPinnedStorenode()
 	if err != nil {
 		m.logger.Error("Could not obtain the pinned storenode", zap.Error(err))
@@ -338,6 +349,29 @@ func (m *StorenodeCycle) GetActiveStorenode() peer.ID {
 	return m.activeStorenode
 }

+func (m *StorenodeCycle) GetActiveStorenodePeerInfo() peer.AddrInfo {
+	m.RLock()
+	defer m.RUnlock()
+
+	if m.storenodeConfigProvider == nil {
+		m.logger.Debug("storenodeConfigProvider not yet setup")
+		return peer.AddrInfo{}
+	}
+
+	storeNodes, err := m.storenodeConfigProvider.Storenodes()
+	if err != nil {
+		return peer.AddrInfo{}
+	}
+
+	for _, p := range storeNodes {
+		if p.ID == m.activeStorenode {
+			return p
+		}
+	}
+
+	return peer.AddrInfo{}
+}
+
 func (m *StorenodeCycle) IsStorenodeAvailable(peerID peer.ID) bool {
 	return m.storenodeStatus(peerID) == connected
 }
@@ -37,7 +37,7 @@ type HistoryRetriever struct {

 type HistoryProcessor interface {
 	OnEnvelope(env *protocol.Envelope, processEnvelopes bool) error
-	OnRequestFailed(requestID []byte, peerID peer.ID, err error)
+	OnRequestFailed(requestID []byte, peerInfo peer.AddrInfo, err error)
 }

 func NewHistoryRetriever(store common.StorenodeRequestor, historyProcessor HistoryProcessor, logger *zap.Logger) *HistoryRetriever {
@@ -51,7 +51,7 @@ func NewHistoryRetriever(store common.StorenodeRequestor, historyProcessor Histo
 func (hr *HistoryRetriever) Query(
 	ctx context.Context,
 	criteria store.FilterCriteria,
-	storenodeID peer.ID,
+	storenode peer.AddrInfo,
 	pageLimit uint64,
 	shouldProcessNextPage func(int) (bool, uint64),
 	processEnvelopes bool,
@@ -178,7 +178,7 @@ loop:
 				newCriteria.TimeStart = timeStart
 				newCriteria.TimeEnd = timeEnd

-				cursor, envelopesCount, err := hr.createMessagesRequest(queryCtx, storenodeID, newCriteria, w.cursor, w.limit, true, processEnvelopes, logger)
+				cursor, envelopesCount, err := hr.createMessagesRequest(queryCtx, storenode, newCriteria, w.cursor, w.limit, true, processEnvelopes, logger)
 				queryCancel()

 				if err != nil {
@@ -241,7 +241,7 @@ loop:

 func (hr *HistoryRetriever) createMessagesRequest(
 	ctx context.Context,
-	peerID peer.ID,
+	peerInfo peer.AddrInfo,
 	criteria store.FilterCriteria,
 	cursor []byte,
 	limit uint64,
@@ -257,7 +257,7 @@ func (hr *HistoryRetriever) createMessagesRequest(
 		})

 		go func() {
-			storeCursor, envelopesCount, err = hr.requestStoreMessages(ctx, peerID, criteria, cursor, limit, processEnvelopes)
+			storeCursor, envelopesCount, err = hr.requestStoreMessages(ctx, peerInfo, criteria, cursor, limit, processEnvelopes)
 			resultCh <- struct {
 				storeCursor    []byte
 				envelopesCount int
@@ -273,7 +273,7 @@ func (hr *HistoryRetriever) createMessagesRequest(
 		}
 	} else {
 		go func() {
-			_, _, err = hr.requestStoreMessages(ctx, peerID, criteria, cursor, limit, false)
+			_, _, err = hr.requestStoreMessages(ctx, peerInfo, criteria, cursor, limit, false)
 			if err != nil {
 				logger.Error("failed to request store messages", zap.Error(err))
 			}
@@ -283,9 +283,9 @@ func (hr *HistoryRetriever) createMessagesRequest(
 	return
 }

-func (hr *HistoryRetriever) requestStoreMessages(ctx context.Context, peerID peer.ID, criteria store.FilterCriteria, cursor []byte, limit uint64, processEnvelopes bool) ([]byte, int, error) {
+func (hr *HistoryRetriever) requestStoreMessages(ctx context.Context, peerInfo peer.AddrInfo, criteria store.FilterCriteria, cursor []byte, limit uint64, processEnvelopes bool) ([]byte, int, error) {
 	requestID := protocol.GenerateRequestID()
-	logger := hr.logger.With(zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", peerID))
+	logger := hr.logger.With(zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", peerInfo.ID))

 	logger.Debug("store.query",
 		logging.Timep("startTime", criteria.TimeStart),
@@ -307,12 +307,12 @@ func (hr *HistoryRetriever) requestStoreMessages(ctx context.Context, peerID pee
 	}

 	queryStart := time.Now()
-	result, err := hr.store.Query(ctx, peerID, storeQueryRequest)
+	result, err := hr.store.Query(ctx, peerInfo, storeQueryRequest)
 	queryDuration := time.Since(queryStart)
 	if err != nil {
 		logger.Error("error querying storenode", zap.Error(err))

-		hr.historyProcessor.OnRequestFailed(requestID, peerID, err)
+		hr.historyProcessor.OnRequestFailed(requestID, peerInfo, err)

 		return nil, 0, err
 	}
@@ -70,7 +70,7 @@ func (h *mockHistoryProcessor) OnEnvelope(env *protocol.Envelope, processEnvelop
 	return nil
 }

-func (h *mockHistoryProcessor) OnRequestFailed(requestID []byte, peerID peer.ID, err error) {
+func (h *mockHistoryProcessor) OnRequestFailed(requestID []byte, peerInfo peer.AddrInfo, err error) {
 }

 func newMockHistoryProcessor() *mockHistoryProcessor {
@@ -92,7 +92,7 @@ func getInitialResponseKey(contentTopics []string) string {
 	return hex.EncodeToString(append([]byte("start"), []byte(contentTopics[0])...))
 }

-func (t *mockStore) Query(ctx context.Context, peerID peer.ID, storeQueryRequest *pb.StoreQueryRequest) (common.StoreRequestResult, error) {
+func (t *mockStore) Query(ctx context.Context, peerInfo peer.AddrInfo, storeQueryRequest *pb.StoreQueryRequest) (common.StoreRequestResult, error) {
 	result := &mockResult{}
 	if len(storeQueryRequest.GetPaginationCursor()) == 0 {
 		initialResponse := getInitialResponseKey(storeQueryRequest.GetContentTopics())
@@ -218,7 +218,7 @@ func TestSuccessBatchExecution(t *testing.T) {
 		ContentFilter: protocol.NewContentFilter("test", topics...),
 	}

-	err = historyRetriever.Query(ctx, criteria, storenodeID, 10, func(i int) (bool, uint64) { return true, 10 }, true)
+	err = historyRetriever.Query(ctx, criteria, peer.AddrInfo{ID: storenodeID}, 10, func(i int) (bool, uint64) { return true, 10 }, true)
 	require.NoError(t, err)
 }

@@ -246,6 +246,6 @@ func TestFailedBatchExecution(t *testing.T) {
 		ContentFilter: protocol.NewContentFilter("test", topics...),
 	}

-	err = historyRetriever.Query(ctx, criteria, storenodeID, 10, func(i int) (bool, uint64) { return true, 10 }, true)
+	err = historyRetriever.Query(ctx, criteria, peer.AddrInfo{ID: storenodeID}, 10, func(i int) (bool, uint64) { return true, 10 }, true)
 	require.Error(t, err)
 }
@@ -10,7 +10,7 @@ import (
 )

 type criteriaInterest struct {
-	peerID        peer.ID
+	peerInfo      peer.AddrInfo
 	contentFilter protocol.ContentFilter
 	lastChecked   time.Time

@@ -19,7 +19,7 @@ type criteriaInterest struct {
 }

 func (c criteriaInterest) equals(other criteriaInterest) bool {
-	if c.peerID != other.peerID {
+	if c.peerInfo.ID != other.peerInfo.ID {
 		return false
 	}

@@ -20,10 +20,10 @@ type defaultStorenodeRequestor struct {
 	store *store.WakuStore
 }

-func (d *defaultStorenodeRequestor) GetMessagesByHash(ctx context.Context, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) (common.StoreRequestResult, error) {
-	return d.store.QueryByHash(ctx, messageHashes, store.WithPeer(peerID), store.WithPaging(false, pageSize))
+func (d *defaultStorenodeRequestor) GetMessagesByHash(ctx context.Context, peerInfo peer.AddrInfo, pageSize uint64, messageHashes []pb.MessageHash) (common.StoreRequestResult, error) {
+	return d.store.QueryByHash(ctx, messageHashes, store.WithPeerAddr(peerInfo.Addrs...), store.WithPaging(false, pageSize))
 }

-func (d *defaultStorenodeRequestor) Query(ctx context.Context, peerID peer.ID, storeQueryRequest *storepb.StoreQueryRequest) (common.StoreRequestResult, error) {
-	return d.store.RequestRaw(ctx, peerID, storeQueryRequest)
+func (d *defaultStorenodeRequestor) Query(ctx context.Context, peerInfo peer.AddrInfo, storeQueryRequest *storepb.StoreQueryRequest) (common.StoreRequestResult, error) {
+	return d.store.RequestRaw(ctx, peerInfo, storeQueryRequest)
 }
@@ -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,16 +66,18 @@ 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),
 	}
 }

-func (m *MissingMessageVerifier) SetCriteriaInterest(peerID peer.ID, contentFilter protocol.ContentFilter) {
+func (m *MissingMessageVerifier) SetCriteriaInterest(peerInfo peer.AddrInfo, contentFilter protocol.ContentFilter) {
 	m.criteriaInterestMu.Lock()
 	defer m.criteriaInterestMu.Unlock()

 	ctx, cancel := context.WithCancel(m.ctx)
 	criteriaInterest := criteriaInterest{
-		peerID:        peerID,
+		peerInfo:      peerInfo,
 		contentFilter: contentFilter,
 		lastChecked:   m.timesource.Now().Add(-m.params.delay),
 		ctx:           ctx,
@@ -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 {
@@ -164,7 +199,7 @@ func (m *MissingMessageVerifier) fetchHistory(c chan<- *protocol.Envelope, inter
 			}

 			m.logger.Error("could not fetch history",
-				zap.Stringer("peerID", interest.peerID),
+				zap.Stringer("peerID", interest.peerInfo.ID),
 				zap.String("pubsubTopic", interest.contentFilter.PubsubTopic),
 				zap.Strings("contentTopics", contentTopics))
 			continue
@@ -207,7 +242,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 	contentTopics := interest.contentFilter.ContentTopics.ToList()

 	logger := m.logger.With(
-		zap.Stringer("peerID", interest.peerID),
+		zap.Stringer("peerID", interest.peerInfo.ID),
 		zap.Strings("contentTopics", contentTopics[batchFrom:batchTo]),
 		zap.String("pubsubTopic", interest.contentFilter.PubsubTopic),
 		logging.Epoch("from", interest.lastChecked),
@@ -226,7 +261,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,

 		return m.storenodeRequestor.Query(
 			ctx,
-			interest.peerID,
+			interest.peerInfo,
 			storeQueryRequest,
 		)
 	}, logger, "retrieving history to check for missing messages")
@@ -309,7 +344,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 					PaginationLimit: proto.Uint64(maxMsgHashesPerRequest),
 				}

-				return m.storenodeRequestor.Query(queryCtx, interest.peerID, storeQueryRequest)
+				return m.storenodeRequestor.Query(queryCtx, interest.peerInfo, storeQueryRequest)
 			}, logger, "retrieving missing messages")
 			if err != nil {
 				if !errors.Is(err, context.Canceled) {
@@ -18,10 +18,10 @@ type defaultStorenodeMessageVerifier struct {
 	store *store.WakuStore
 }

-func (d *defaultStorenodeMessageVerifier) MessageHashesExist(ctx context.Context, requestID []byte, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error) {
+func (d *defaultStorenodeMessageVerifier) MessageHashesExist(ctx context.Context, requestID []byte, peerID peer.AddrInfo, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error) {
 	var opts []store.RequestOption
 	opts = append(opts, store.WithRequestID(requestID))
-	opts = append(opts, store.WithPeer(peerID))
+	opts = append(opts, store.WithPeerAddr(peerID.Addrs...))
 	opts = append(opts, store.WithPaging(false, pageSize))
 	opts = append(opts, store.IncludeData(false))

@@ -33,7 +33,7 @@ type ISentCheck interface {

 type StorenodeMessageVerifier interface {
 	// MessagesExist returns a list of the messages it found from a list of message hashes
-	MessageHashesExist(ctx context.Context, requestID []byte, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error)
+	MessageHashesExist(ctx context.Context, requestID []byte, peerInfo peer.AddrInfo, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error)
 }

 // MessageSentCheck tracks the outgoing messages and check against store node
@@ -211,8 +211,8 @@ func (m *MessageSentCheck) Start() {
 }

 func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []common.Hash, relayTime []uint32, pubsubTopic string) []common.Hash {
-	selectedPeer := m.storenodeCycle.GetActiveStorenode()
-	if selectedPeer == "" {
+	selectedPeer := m.storenodeCycle.GetActiveStorenodePeerInfo()
+	if selectedPeer.ID == "" {
 		m.logger.Error("no store peer id available", zap.String("pubsubTopic", pubsubTopic))
 		return []common.Hash{}
 	}
@@ -224,13 +224,13 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c
 		messageHashes[i] = pb.ToMessageHash(hash.Bytes())
 	}

-	m.logger.Debug("store.queryByHash request", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer), zap.Stringers("messageHashes", messageHashes))
+	m.logger.Debug("store.queryByHash request", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer.ID), zap.Stringers("messageHashes", messageHashes))

 	queryCtx, cancel := context.WithTimeout(ctx, m.storeQueryTimeout)
 	defer cancel()
 	result, err := m.messageVerifier.MessageHashesExist(queryCtx, requestID, selectedPeer, m.maxHashQueryLength, messageHashes)
 	if err != nil {
-		m.logger.Error("store.queryByHash failed", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer), zap.Error(err))
+		m.logger.Error("store.queryByHash failed", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer.ID), zap.Error(err))
 		return []common.Hash{}
 	}

@@ -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)
+}
@@ -79,7 +79,7 @@ func TestPeriodicKeepAlive(t *testing.T) {

 	node2MAddr, err := multiaddr.NewMultiaddr(host1.Addrs()[0].String() + "/p2p/" + host1.ID().String())
 	require.NoError(t, err)
-	_, err = wakuNode.AddPeer(node2MAddr, wps.Static, []string{"waku/rs/1/1"})
+	_, err = wakuNode.AddPeer([]multiaddr.Multiaddr{node2MAddr}, wps.Static, []string{"waku/rs/1/1"})
 	require.NoError(t, err)

 	time.Sleep(time.Second * 2)
@@ -703,8 +703,8 @@ func (w *WakuNode) startStore(ctx context.Context, sub *relay.Subscription) erro

 // AddPeer is used to add a peer and the protocols it support to the node peerstore
 // TODO: Need to update this for autosharding, to only take contentTopics and optional pubSubTopics or provide an alternate API only for contentTopics.
-func (w *WakuNode) AddPeer(address ma.Multiaddr, origin wps.Origin, pubSubTopics []string, protocols ...protocol.ID) (peer.ID, error) {
-	pData, err := w.peermanager.AddPeer(address, origin, pubSubTopics, protocols...)
+func (w *WakuNode) AddPeer(addresses []ma.Multiaddr, origin wps.Origin, pubSubTopics []string, protocols ...protocol.ID) (peer.ID, error) {
+	pData, err := w.peermanager.AddPeer(addresses, origin, pubSubTopics, protocols...)
 	if err != nil {
 		return "", err
 	}
@@ -268,7 +268,7 @@ func TestDecoupledStoreFromRelay(t *testing.T) {
 	require.NoError(t, err)
 	defer wakuNode2.Stop()

-	peerID, err := wakuNode2.AddPeer(wakuNode1.ListenAddresses()[0], peerstore.Static, []string{relay.DefaultWakuTopic}, filter.FilterSubscribeID_v20beta1)
+	peerID, err := wakuNode2.AddPeer(wakuNode1.ListenAddresses(), peerstore.Static, []string{relay.DefaultWakuTopic}, filter.FilterSubscribeID_v20beta1)
 	require.NoError(t, err)

 	subscription, err := wakuNode2.FilterLightnode().Subscribe(ctx, protocol.ContentFilter{
@@ -317,7 +317,7 @@ func TestDecoupledStoreFromRelay(t *testing.T) {
 	require.NoError(t, err)
 	defer wakuNode3.Stop()

-	_, err = wakuNode3.AddPeer(wakuNode2.ListenAddresses()[0], peerstore.Static, []string{relay.DefaultWakuTopic}, legacy_store.StoreID_v20beta4)
+	_, err = wakuNode3.AddPeer(wakuNode2.ListenAddresses(), peerstore.Static, []string{relay.DefaultWakuTopic}, legacy_store.StoreID_v20beta4)
 	require.NoError(t, err)
 	time.Sleep(2 * time.Second)
 	// NODE2 should have returned the message received via filter
@@ -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()
 		}
 	}
 }
@@ -684,13 +678,19 @@ func AddrInfoToPeerData(origin wps.Origin, peerID peer.ID, host host.Host, pubsu
 }

 // AddPeer adds peer to the peerStore and also to service slots
-func (pm *PeerManager) AddPeer(address ma.Multiaddr, origin wps.Origin, pubsubTopics []string, protocols ...protocol.ID) (*service.PeerData, error) {
+func (pm *PeerManager) AddPeer(addresses []ma.Multiaddr, origin wps.Origin, pubsubTopics []string, protocols ...protocol.ID) (*service.PeerData, error) {
 	//Assuming all addresses have peerId
-	info, err := peer.AddrInfoFromP2pAddr(address)
+	infoArr, err := peer.AddrInfosFromP2pAddrs(addresses...)
 	if err != nil {
 		return nil, err
 	}

+	if len(infoArr) > 1 {
+		return nil, errors.New("only a single peerID is expected in AddPeer")
+	}
+
+	info := infoArr[0]
+
 	//Add Service peers to serviceSlots.
 	for _, proto := range protocols {
 		pm.addPeerToServiceSlot(proto, info.ID)
@@ -703,11 +703,8 @@ func (pm *PeerManager) AddPeer(address ma.Multiaddr, origin wps.Origin, pubsubTo
 	}

 	pData := &service.PeerData{
-		Origin: origin,
-		AddrInfo: peer.AddrInfo{
-			ID:    info.ID,
-			Addrs: info.Addrs,
-		},
+		Origin:       origin,
+		AddrInfo:     info,
 		PubsubTopics: pubsubTopics,
 	}

@@ -749,6 +746,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 +760,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)
-	}
 }
@@ -12,6 +12,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 	libp2pProtocol "github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/libp2p/go-libp2p/p2p/host/peerstore/pstoremem"
+	"github.com/multiformats/go-multiaddr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 	"github.com/waku-org/go-waku/tests"
@@ -57,7 +58,7 @@ func TestServiceSlots(t *testing.T) {

 	// add h2 peer to peer manager
 	t.Log(h2.ID())
-	_, err = pm.AddPeer(tests.GetAddr(h2), wps.Static, []string{""}, libp2pProtocol.ID(protocol))
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h2)}, wps.Static, []string{""}, libp2pProtocol.ID(protocol))
 	require.NoError(t, err)

 	///////////////
@@ -71,7 +72,7 @@ func TestServiceSlots(t *testing.T) {
 	require.Equal(t, h2.ID(), peers[0])

 	// add h3 peer to peer manager
-	_, err = pm.AddPeer(tests.GetAddr(h3), wps.Static, []string{""}, libp2pProtocol.ID(protocol))
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h3)}, wps.Static, []string{""}, libp2pProtocol.ID(protocol))
 	require.NoError(t, err)

 	// check that returned peer is h2 or h3 peer
@@ -96,7 +97,7 @@ func TestServiceSlots(t *testing.T) {
 	require.Error(t, err, utils.ErrNoPeersAvailable)

 	// add h4 peer for protocol1
-	_, err = pm.AddPeer(tests.GetAddr(h4), wps.Static, []string{""}, libp2pProtocol.ID(protocol1))
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h4)}, wps.Static, []string{""}, libp2pProtocol.ID(protocol1))
 	require.NoError(t, err)

 	//Test peer selection for protocol1
@@ -124,10 +125,10 @@ func TestPeerSelection(t *testing.T) {
 	defer h3.Close()

 	protocol := libp2pProtocol.ID("test/protocol")
-	_, err = pm.AddPeer(tests.GetAddr(h2), wps.Static, []string{"/waku/2/rs/2/1", "/waku/2/rs/2/2"}, libp2pProtocol.ID(protocol))
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h2)}, wps.Static, []string{"/waku/2/rs/2/1", "/waku/2/rs/2/2"}, libp2pProtocol.ID(protocol))
 	require.NoError(t, err)

-	_, err = pm.AddPeer(tests.GetAddr(h3), wps.Static, []string{"/waku/2/rs/2/1"}, libp2pProtocol.ID(protocol))
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h3)}, wps.Static, []string{"/waku/2/rs/2/1"}, libp2pProtocol.ID(protocol))
 	require.NoError(t, err)

 	_, err = pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: protocol})
@@ -158,7 +159,7 @@ func TestPeerSelection(t *testing.T) {
 	h4, err := tests.MakeHost(ctx, 0, rand.Reader)
 	require.NoError(t, err)
 	defer h4.Close()
-	_, err = pm.AddPeer(tests.GetAddr(h4), wps.Static, []string{"/waku/2/rs/2/1"}, libp2pProtocol.ID(protocol))
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h4)}, wps.Static, []string{"/waku/2/rs/2/1"}, libp2pProtocol.ID(protocol))
 	require.NoError(t, err)

 	peerIDs, err = pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: protocol, PubsubTopics: []string{"/waku/2/rs/2/1"}, MaxPeers: 3})
@@ -185,7 +186,7 @@ func TestDefaultProtocol(t *testing.T) {
 	defer h5.Close()

 	//Test peer selection for relay protocol from peer store
-	_, err = pm.AddPeer(tests.GetAddr(h5), wps.Static, []string{""}, relay.WakuRelayID_v200)
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h5)}, wps.Static, []string{""}, relay.WakuRelayID_v200)
 	require.NoError(t, err)

 	// since we are not passing peerList, selectPeer fn using filterByProto checks in PeerStore for peers with same protocol.
@@ -206,7 +207,7 @@ func TestAdditionAndRemovalOfPeer(t *testing.T) {
 	require.NoError(t, err)
 	defer h6.Close()

-	_, err = pm.AddPeer(tests.GetAddr(h6), wps.Static, []string{""}, protocol2)
+	_, err = pm.AddPeer([]multiaddr.Multiaddr{tests.GetAddr(h6)}, wps.Static, []string{""}, protocol2)
 	require.NoError(t, err)

 	peers, err := pm.SelectPeers(PeerSelectionCriteria{SelectionType: Automatic, Proto: protocol2})
@@ -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,7 +15,9 @@ 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/multiformats/go-multiaddr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/waku-org/go-waku/logging"
 	"github.com/waku-org/go-waku/waku/v2/onlinechecker"
@@ -54,6 +56,7 @@ type WakuFilterLightNode struct {
 	log              *zap.Logger
 	subscriptions    *subscription.SubscriptionsMap
 	pm               *peermanager.PeerManager
+	limiter          *utils.RateLimiter
 	peerPingInterval time.Duration
 }

@@ -89,6 +92,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 +103,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 +167,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 +269,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 +311,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")
@@ -325,7 +349,7 @@ func (wf *WakuFilterLightNode) handleFilterSubscribeOptions(ctx context.Context,

 	//Add Peer to peerstore.
 	if params.pm != nil && params.peerAddr != nil {
-		pData, err := wf.pm.AddPeer(params.peerAddr, peerstore.Static, maps.Keys(pubSubTopicMap), FilterSubscribeID_v20beta1)
+		pData, err := wf.pm.AddPeer([]multiaddr.Multiaddr{params.peerAddr}, peerstore.Static, maps.Keys(pubSubTopicMap), FilterSubscribeID_v20beta1)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -337,7 +361,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 +374,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,7 +24,7 @@ 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 {
+	if err != nil && wf.onlineChecker.IsOnline() {
 		wf.log.Warn("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.
@@ -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)

@@ -10,6 +10,7 @@ import (

 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/multiformats/go-multiaddr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/suite"
 	"github.com/waku-org/go-waku/tests"
@@ -22,6 +23,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 {
@@ -101,7 +103,7 @@ func (s *FilterTestSuite) TearDownTest() {

 func (s *FilterTestSuite) ConnectToFullNode(h1 *WakuFilterLightNode, h2 *WakuFilterFullNode) {
 	mAddr := tests.GetAddr(h2.h)
-	_, err := h1.pm.AddPeer(mAddr, wps.Static, []string{s.TestTopic}, FilterSubscribeID_v20beta1)
+	_, err := h1.pm.AddPeer([]multiaddr.Multiaddr{mAddr}, wps.Static, []string{s.TestTopic}, FilterSubscribeID_v20beta1)
 	s.Log.Info("add peer", zap.Stringer("mAddr", mAddr))
 	s.Require().NoError(err)
 }
@@ -133,7 +135,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 +168,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}
@@ -310,7 +310,7 @@ func (store *WakuStore) Query(ctx context.Context, query Query, opts ...HistoryR

 		//Add Peer to peerstore.
 		if store.pm != nil && params.peerAddr != nil {
-			pData, err := store.pm.AddPeer(params.peerAddr, peerstore.Static, pubsubTopics, StoreID_v20beta4)
+			pData, err := store.pm.AddPeer([]multiaddr.Multiaddr{params.peerAddr}, peerstore.Static, pubsubTopics, StoreID_v20beta4)
 			if err != nil {
 				return nil, err
 			}
@@ -13,7 +13,9 @@ 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/multiformats/go-multiaddr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/waku-org/go-waku/logging"
 	"github.com/waku-org/go-waku/waku/v2/peermanager"
@@ -24,7 +26,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 +41,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 +60,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 +108,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 +200,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
 	}
@@ -273,7 +279,7 @@ func (wakuLP *WakuLightPush) handleOpts(ctx context.Context, message *wpb.WakuMe
 	}

 	if params.pm != nil && params.peerAddr != nil {
-		pData, err := wakuLP.pm.AddPeer(params.peerAddr, peerstore.Static, []string{params.pubsubTopic}, LightPushID_v20beta1)
+		pData, err := wakuLP.pm.AddPeer([]multiaddr.Multiaddr{params.peerAddr}, peerstore.Static, []string{params.pubsubTopic}, LightPushID_v20beta1)
 		if err != nil {
 			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)
@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-msgio/pbio"
+	"github.com/multiformats/go-multiaddr"
 	"github.com/waku-org/go-waku/waku/v2/peermanager"
 	"github.com/waku-org/go-waku/waku/v2/peerstore"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
@@ -36,7 +37,7 @@ func (wakuPX *WakuPeerExchange) Request(ctx context.Context, numPeers int, opts
 	}

 	if params.pm != nil && params.peerAddr != nil {
-		pData, err := wakuPX.pm.AddPeer(params.peerAddr, peerstore.Static, []string{}, PeerExchangeID_v20alpha1)
+		pData, err := wakuPX.pm.AddPeer([]multiaddr.Multiaddr{params.peerAddr}, peerstore.Static, []string{}, PeerExchangeID_v20alpha1)
 		if err != nil {
 			return 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
@@ -194,15 +196,15 @@ func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...Requ
 	return result, nil
 }

-func (s *WakuStore) RequestRaw(ctx context.Context, peerID peer.ID, storeRequest *pb.StoreQueryRequest) (Result, error) {
+func (s *WakuStore) RequestRaw(ctx context.Context, peerInfo peer.AddrInfo, storeRequest *pb.StoreQueryRequest) (Result, error) {
 	err := storeRequest.Validate()
 	if err != nil {
 		return nil, err
 	}

 	var params Parameters
-	params.selectedPeer = peerID
-	if params.selectedPeer == "" {
+	params.peerAddr = peerInfo.Addrs
+	if len(params.peerAddr) == 0 {
 		return nil, ErrMustSelectPeer
 	}

@@ -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
@@ -11,7 +11,7 @@ import (

 type Parameters struct {
 	selectedPeer      peer.ID
-	peerAddr          multiaddr.Multiaddr
+	peerAddr          []multiaddr.Multiaddr
 	peerSelectionType peermanager.PeerSelection
 	preferredPeers    peer.IDSlice
 	requestID         []byte
@@ -33,7 +33,7 @@ type RequestOption func(*Parameters) error
 func WithPeer(p peer.ID) RequestOption {
 	return func(params *Parameters) error {
 		params.selectedPeer = p
-		if params.peerAddr != nil {
+		if len(params.peerAddr) != 0 {
 			return errors.New("WithPeer and WithPeerAddr options are mutually exclusive")
 		}
 		return nil
@@ -43,7 +43,7 @@ func WithPeer(p peer.ID) RequestOption {
 // WithPeerAddr is an option used to specify a peerAddress to request the message history.
 // This new peer will be added to peerStore.
 // Note that this option is mutually exclusive to WithPeerAddr, only one of them can be used.
-func WithPeerAddr(pAddr multiaddr.Multiaddr) RequestOption {
+func WithPeerAddr(pAddr ...multiaddr.Multiaddr) RequestOption {
 	return func(params *Parameters) error {
 		params.peerAddr = pAddr
 		if params.selectedPeer != "" {
@@ -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
Richard Ramos	bf357f223b	refactor: use peerInfo instead of peerID	2025-01-22 14:21:17 -04:00
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