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
61 changed files with 2216 additions and 212 deletions
@@ -309,11 +309,6 @@ func Execute(options NodeOptions) error {
 		nodeOpts = append(nodeOpts, discv5Opts)
 	}

-	wakuNode, err := node.New(nodeOpts...)
-	if err != nil {
-		return fmt.Errorf("could not instantiate waku: %w", err)
-	}
-
 	//Process pubSub and contentTopics specified and arrive at all corresponding pubSubTopics
 	pubSubTopicMap, err := processTopics(options)
 	if err != nil {
@@ -325,6 +320,22 @@ func Execute(options NodeOptions) error {
 		pubSubTopicMapKeys = append(pubSubTopicMapKeys, k)
 	}

+	rs, err := wprotocol.TopicsToRelayShards(pubSubTopicMapKeys...)
+	if err == nil {
+		if len(rs) == 1 {
+			nodeOpts = append(nodeOpts, node.WithShards(rs[0].ShardIDs))
+		} else {
+			logger.Warn("could not set ENR shard info", zap.String("error", "invalid number of clusters found"), zap.Int("numClusters", len(rs)))
+		}
+	} else {
+		logger.Warn("could not obtain list of shards", zap.Error(err))
+	}
+
+	wakuNode, err := node.New(nodeOpts...)
+	if err != nil {
+		return fmt.Errorf("could not instantiate waku: %w", err)
+	}
+
 	if err = wakuNode.Start(ctx); err != nil {
 		return nonRecoverError(err)
 	}
@@ -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=
@@ -74,6 +74,14 @@ func (t timestamp) String() string {
 	return time.Unix(0, int64(t)).Format(time.RFC3339)
 }

+func Timep(key string, time *int64) zapcore.Field {
+	if time == nil {
+		return zap.String(key, "-")
+	} else {
+		return Time(key, *time)
+	}
+}
+
 func Epoch(key string, time time.Time) zap.Field {
 	return zap.String(key, fmt.Sprintf("%d", time.UnixNano()))
 }
@@ -0,0 +1,39 @@
+package common
+
+import (
+	"context"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/peerstore"
+	"github.com/libp2p/go-libp2p/p2p/protocol/ping"
+)
+
+type Pinger interface {
+	PingPeer(ctx context.Context, peerInfo peer.AddrInfo) (time.Duration, error)
+}
+
+type defaultPingImpl struct {
+	host host.Host
+}
+
+func NewDefaultPinger(host host.Host) Pinger {
+	return &defaultPingImpl{
+		host: host,
+	}
+}
+
+func (d *defaultPingImpl) PingPeer(ctx context.Context, peerInfo peer.AddrInfo) (time.Duration, error) {
+	d.host.Peerstore().AddAddrs(peerInfo.ID, peerInfo.Addrs, peerstore.AddressTTL)
+	pingResultCh := ping.Ping(ctx, d.host, peerInfo.ID)
+	select {
+	case <-ctx.Done():
+		return 0, ctx.Err()
+	case r := <-pingResultCh:
+		if r.Error != nil {
+			return 0, r.Error
+		}
+		return r.RTT, nil
+	}
+}
@@ -0,0 +1,17 @@
+package common
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
+)
+
+type StoreRequestResult interface {
+	Cursor() []byte
+	IsComplete() bool
+	PeerID() peer.ID
+	Next(ctx context.Context, opts ...store.RequestOption) error // TODO: see how to decouple store.RequestOption
+	Messages() []*pb.WakuMessageKeyValue
+}
@@ -0,0 +1,12 @@
+package common
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
+)
+
+type StorenodeRequestor interface {
+	Query(ctx context.Context, peerID peer.ID, 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)
@@ -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)
 }

@@ -230,6 +299,7 @@ func (mgr *FilterManager) UnsubscribeFilter(filterID string) {
 		}
 		if len(af.sub.ContentFilter.ContentTopics) == 0 {
 			af.cancel()
+			delete(mgr.filterSubscriptions, filterConfig.ID)
 		} else {
 			go af.sub.Unsubscribe(filterConfig.contentFilter)
 		}
@@ -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)

@@ -0,0 +1,526 @@
+package history
+
+import (
+	"context"
+	"crypto/rand"
+	"errors"
+	"fmt"
+	"math"
+	"math/big"
+	"net"
+	"net/http"
+	"runtime"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/api/common"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+	"go.uber.org/zap"
+)
+
+const defaultBackoff = 10 * time.Second
+const graylistBackoff = 3 * time.Minute
+const storenodeVerificationInterval = time.Second
+const storenodeMaxFailedRequests uint = 2
+const minStorenodesToChooseFrom = 3
+const isAndroidEmulator = runtime.GOOS == "android" && runtime.GOARCH == "amd64"
+const findNearestMailServer = !isAndroidEmulator
+const overrideDNS = runtime.GOOS == "android" || runtime.GOOS == "ios"
+const bootstrapDNS = "8.8.8.8:53"
+
+type connStatus int
+
+const (
+	disconnected connStatus = iota + 1
+	connected
+)
+
+type peerStatus struct {
+	status                connStatus
+	canConnectAfter       time.Time
+	lastConnectionAttempt time.Time
+}
+
+type StorenodeConfigProvider interface {
+	UseStorenodes() (bool, error)
+	GetPinnedStorenode() (peer.AddrInfo, error)
+	Storenodes() ([]peer.AddrInfo, error)
+}
+
+type StorenodeCycle struct {
+	sync.RWMutex
+
+	logger *zap.Logger
+
+	storenodeConfigProvider StorenodeConfigProvider
+	pinger                  common.Pinger
+
+	StorenodeAvailableOneshotEmitter *OneShotEmitter[struct{}]
+	StorenodeChangedEmitter          *Emitter[peer.ID]
+	StorenodeNotWorkingEmitter       *Emitter[struct{}]
+	StorenodeAvailableEmitter        *Emitter[peer.ID]
+
+	failedRequests map[peer.ID]uint
+
+	peersMutex      sync.RWMutex
+	activeStorenode peer.ID
+	peers           map[peer.ID]peerStatus
+}
+
+func NewStorenodeCycle(logger *zap.Logger, pinger common.Pinger) *StorenodeCycle {
+	return &StorenodeCycle{
+		StorenodeAvailableOneshotEmitter: NewOneshotEmitter[struct{}](),
+		StorenodeChangedEmitter:          NewEmitter[peer.ID](),
+		StorenodeNotWorkingEmitter:       NewEmitter[struct{}](),
+		StorenodeAvailableEmitter:        NewEmitter[peer.ID](),
+		pinger:                           pinger,
+		logger:                           logger.Named("storenode-cycle"),
+	}
+}
+
+func (m *StorenodeCycle) Start(ctx context.Context) {
+	m.logger.Debug("starting storenode cycle")
+	m.failedRequests = make(map[peer.ID]uint)
+	m.peers = make(map[peer.ID]peerStatus)
+
+	go m.verifyStorenodeStatus(ctx)
+}
+
+func (m *StorenodeCycle) DisconnectActiveStorenode(backoff time.Duration) {
+	m.Lock()
+	defer m.Unlock()
+
+	m.disconnectActiveStorenode(backoff)
+}
+
+func (m *StorenodeCycle) connectToNewStorenodeAndWait(ctx context.Context) error {
+	// Handle pinned storenodes
+	m.logger.Info("disconnecting storenode")
+	pinnedStorenode, err := m.storenodeConfigProvider.GetPinnedStorenode()
+	if err != nil {
+		m.logger.Error("could not obtain the pinned storenode", zap.Error(err))
+		return err
+	}
+
+	// If no pinned storenode, no need to disconnect and wait for it to be available
+	if pinnedStorenode.ID == "" {
+		m.disconnectActiveStorenode(graylistBackoff)
+	}
+
+	return m.findNewStorenode(ctx)
+}
+
+func (m *StorenodeCycle) disconnectStorenode(backoffDuration time.Duration) error {
+	if m.activeStorenode == "" {
+		m.logger.Info("no active storenode")
+		return nil
+	}
+
+	m.logger.Info("disconnecting active storenode", zap.Stringer("peerID", m.activeStorenode))
+
+	m.peersMutex.Lock()
+	pInfo, ok := m.peers[m.activeStorenode]
+	if ok {
+		pInfo.status = disconnected
+		pInfo.canConnectAfter = time.Now().Add(backoffDuration)
+		m.peers[m.activeStorenode] = pInfo
+	} else {
+		m.peers[m.activeStorenode] = peerStatus{
+			status:          disconnected,
+			canConnectAfter: time.Now().Add(backoffDuration),
+		}
+	}
+	m.peersMutex.Unlock()
+
+	m.activeStorenode = ""
+
+	return nil
+}
+
+func (m *StorenodeCycle) disconnectActiveStorenode(backoffDuration time.Duration) {
+	err := m.disconnectStorenode(backoffDuration)
+	if err != nil {
+		m.logger.Error("failed to disconnect storenode", zap.Error(err))
+	}
+
+	m.StorenodeChangedEmitter.Emit("")
+}
+
+func (m *StorenodeCycle) Cycle(ctx context.Context) {
+	if m.storenodeConfigProvider == nil {
+		m.logger.Debug("storenodeConfigProvider not yet setup")
+		return
+	}
+
+	m.logger.Info("Automatically switching storenode")
+
+	if m.activeStorenode != "" {
+		m.disconnectActiveStorenode(graylistBackoff)
+	}
+
+	useStorenode, err := m.storenodeConfigProvider.UseStorenodes()
+	if err != nil {
+		m.logger.Error("failed to get use storenodes", zap.Error(err))
+		return
+	}
+
+	if !useStorenode {
+		m.logger.Info("Skipping storenode search due to useStorenode being false")
+		return
+	}
+
+	err = m.findNewStorenode(ctx)
+	if err != nil {
+		m.logger.Error("Error getting new storenode", zap.Error(err))
+	}
+}
+
+func poolSize(fleetSize int) int {
+	return int(math.Ceil(float64(fleetSize) / 4))
+}
+
+func (m *StorenodeCycle) getAvailableStorenodesSortedByRTT(ctx context.Context, allStorenodes []peer.AddrInfo) []peer.AddrInfo {
+	peerIDToInfo := make(map[peer.ID]peer.AddrInfo)
+	for _, p := range allStorenodes {
+		peerIDToInfo[p.ID] = p
+	}
+
+	availableStorenodes := make(map[peer.ID]time.Duration)
+	availableStorenodesMutex := sync.Mutex{}
+	availableStorenodesWg := sync.WaitGroup{}
+	for _, storenode := range allStorenodes {
+		availableStorenodesWg.Add(1)
+		go func(peerInfo peer.AddrInfo) {
+			defer availableStorenodesWg.Done()
+			ctx, cancel := context.WithTimeout(ctx, 4*time.Second)
+			defer cancel()
+
+			rtt, err := m.pinger.PingPeer(ctx, peerInfo)
+			if err == nil { // pinging storenodes might fail, but we don't care
+				availableStorenodesMutex.Lock()
+				availableStorenodes[peerInfo.ID] = rtt
+				availableStorenodesMutex.Unlock()
+			}
+		}(storenode)
+	}
+	availableStorenodesWg.Wait()
+
+	if len(availableStorenodes) == 0 {
+		m.logger.Warn("No storenodes available") // Do nothing..
+		return nil
+	}
+
+	var sortedStorenodes []SortedStorenode
+	for storenodeID, rtt := range availableStorenodes {
+		sortedStorenode := SortedStorenode{
+			Storenode: peerIDToInfo[storenodeID],
+			RTT:       rtt,
+		}
+		m.peersMutex.Lock()
+		pInfo, ok := m.peers[storenodeID]
+		m.peersMutex.Unlock()
+		if ok && time.Now().Before(pInfo.canConnectAfter) {
+			continue // We can't connect to this node yet
+		}
+		sortedStorenodes = append(sortedStorenodes, sortedStorenode)
+	}
+	sort.Sort(byRTTMsAndCanConnectBefore(sortedStorenodes))
+
+	result := make([]peer.AddrInfo, len(sortedStorenodes))
+	for i, s := range sortedStorenodes {
+		result[i] = s.Storenode
+	}
+
+	return result
+}
+
+func (m *StorenodeCycle) findNewStorenode(ctx context.Context) error {
+	// we have to override DNS manually because of https://github.com/status-im/status-mobile/issues/19581
+	if overrideDNS {
+		var dialer net.Dialer
+		net.DefaultResolver = &net.Resolver{
+			PreferGo: false,
+			Dial: func(context context.Context, _, _ string) (net.Conn, error) {
+				conn, err := dialer.DialContext(context, "udp", bootstrapDNS)
+				if err != nil {
+					return nil, err
+				}
+				return conn, nil
+			},
+		}
+	}
+
+	pinnedStorenode, err := m.storenodeConfigProvider.GetPinnedStorenode()
+	if err != nil {
+		m.logger.Error("Could not obtain the pinned storenode", zap.Error(err))
+		return err
+	}
+
+	if pinnedStorenode.ID != "" {
+		return m.setActiveStorenode(pinnedStorenode.ID)
+	}
+
+	m.logger.Info("Finding a new storenode..")
+
+	allStorenodes, err := m.storenodeConfigProvider.Storenodes()
+	if err != nil {
+		return err
+	}
+
+	//	TODO: remove this check once sockets are stable on x86_64 emulators
+	if findNearestMailServer {
+		allStorenodes = m.getAvailableStorenodesSortedByRTT(ctx, allStorenodes)
+	}
+
+	// Picks a random storenode amongs the ones with the lowest latency
+	// The pool size is 1/4 of the storenodes were pinged successfully
+	// If the pool size is less than `minStorenodesToChooseFrom`, it will
+	// pick a storenode fromm all the available storenodes
+	pSize := poolSize(len(allStorenodes) - 1)
+	if pSize <= minStorenodesToChooseFrom {
+		pSize = len(allStorenodes)
+		if pSize <= 0 {
+			m.logger.Warn("No storenodes available") // Do nothing..
+			return nil
+		}
+	}
+
+	r, err := rand.Int(rand.Reader, big.NewInt(int64(pSize)))
+	if err != nil {
+		return err
+	}
+
+	ms := allStorenodes[r.Int64()]
+	return m.setActiveStorenode(ms.ID)
+}
+
+func (m *StorenodeCycle) storenodeStatus(peerID peer.ID) connStatus {
+	m.peersMutex.RLock()
+	defer m.peersMutex.RUnlock()
+
+	peer, ok := m.peers[peerID]
+	if !ok {
+		return disconnected
+	}
+	return peer.status
+}
+
+func (m *StorenodeCycle) setActiveStorenode(peerID peer.ID) error {
+	m.activeStorenode = peerID
+
+	m.StorenodeChangedEmitter.Emit(m.activeStorenode)
+
+	storenodeStatus := m.storenodeStatus(peerID)
+	if storenodeStatus != connected {
+		m.peersMutex.Lock()
+		m.peers[peerID] = peerStatus{
+			status:                connected,
+			lastConnectionAttempt: time.Now(),
+			canConnectAfter:       time.Now().Add(defaultBackoff),
+		}
+		m.peersMutex.Unlock()
+
+		m.failedRequests[peerID] = 0
+		m.logger.Info("storenode available", zap.Stringer("peerID", m.activeStorenode))
+
+		m.StorenodeAvailableOneshotEmitter.Emit(struct{}{}) // Maybe can be refactored away?
+		m.StorenodeAvailableEmitter.Emit(m.activeStorenode)
+	}
+	return nil
+}
+
+func (m *StorenodeCycle) GetActiveStorenode() peer.ID {
+	m.RLock()
+	defer m.RUnlock()
+
+	return m.activeStorenode
+}
+
+func (m *StorenodeCycle) IsStorenodeAvailable(peerID peer.ID) bool {
+	return m.storenodeStatus(peerID) == connected
+}
+
+func (m *StorenodeCycle) penalizeStorenode(id peer.ID) {
+	m.peersMutex.Lock()
+	defer m.peersMutex.Unlock()
+	pInfo, ok := m.peers[id]
+	if !ok {
+		pInfo.status = disconnected
+	}
+
+	pInfo.canConnectAfter = time.Now().Add(graylistBackoff)
+	m.peers[id] = pInfo
+}
+
+func (m *StorenodeCycle) verifyStorenodeStatus(ctx context.Context) {
+	ticker := time.NewTicker(storenodeVerificationInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			err := m.disconnectStorenodeIfRequired(ctx)
+			if err != nil {
+				m.logger.Error("failed to handle storenode cycle event", zap.Error(err))
+				continue
+			}
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (m *StorenodeCycle) disconnectStorenodeIfRequired(ctx context.Context) error {
+	m.logger.Debug("wakuV2 storenode status verification")
+
+	if m.activeStorenode == "" {
+		// No active storenode, find a new one
+		m.Cycle(ctx)
+		return nil
+	}
+
+	// Check whether we want to disconnect the active storenode
+	if m.failedRequests[m.activeStorenode] >= storenodeMaxFailedRequests {
+		m.penalizeStorenode(m.activeStorenode)
+		m.StorenodeNotWorkingEmitter.Emit(struct{}{})
+
+		m.logger.Info("too many failed requests", zap.Stringer("storenode", m.activeStorenode))
+		m.failedRequests[m.activeStorenode] = 0
+		return m.connectToNewStorenodeAndWait(ctx)
+	}
+
+	return nil
+}
+
+func (m *StorenodeCycle) SetStorenodeConfigProvider(provider StorenodeConfigProvider) {
+	m.storenodeConfigProvider = provider
+}
+
+func (m *StorenodeCycle) WaitForAvailableStoreNode(ctx context.Context) bool {
+	// Note: Add 1 second to timeout, because the storenode cycle has 1 second ticker, which doesn't tick on start.
+	// This can be improved after merging https://github.com/status-im/status-go/pull/4380.
+	// NOTE: https://stackoverflow.com/questions/32705582/how-to-get-time-tick-to-tick-immediately
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for !m.IsStorenodeAvailable(m.activeStorenode) {
+			select {
+			case <-m.StorenodeAvailableOneshotEmitter.Subscribe():
+			case <-ctx.Done():
+				if errors.Is(ctx.Err(), context.Canceled) {
+					return
+				}
+
+				// Wait for an additional second, but handle cancellation
+				select {
+				case <-time.After(1 * time.Second):
+				case <-ctx.Done(): // context was cancelled
+				}
+
+				return
+
+			}
+		}
+	}()
+
+	select {
+	case <-waitForWaitGroup(&wg):
+	case <-ctx.Done():
+		// Wait for an additional second, but handle cancellation
+		select {
+		case <-time.After(1 * time.Second):
+		case <-ctx.Done(): // context was cancelled o
+		}
+	}
+
+	return m.IsStorenodeAvailable(m.activeStorenode)
+}
+
+func waitForWaitGroup(wg *sync.WaitGroup) <-chan struct{} {
+	ch := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(ch)
+	}()
+	return ch
+}
+
+type storenodeTaskParameters struct {
+	customPeerID peer.ID
+}
+
+type StorenodeTaskOption func(*storenodeTaskParameters)
+
+func WithPeerID(peerID peer.ID) StorenodeTaskOption {
+	return func(stp *storenodeTaskParameters) {
+		stp.customPeerID = peerID
+	}
+}
+
+func (m *StorenodeCycle) PerformStorenodeTask(fn func() error, options ...StorenodeTaskOption) error {
+	params := storenodeTaskParameters{}
+	for _, opt := range options {
+		opt(&params)
+	}
+
+	peerID := params.customPeerID
+	if peerID == "" {
+		peerID = m.GetActiveStorenode()
+	}
+
+	if peerID == "" {
+		return errors.New("storenode not available")
+	}
+
+	m.RLock()
+	defer m.RUnlock()
+
+	var tries uint = 0
+	for tries < storenodeMaxFailedRequests {
+		if params.customPeerID == "" && m.storenodeStatus(peerID) != connected {
+			return errors.New("storenode not available")
+		}
+		m.logger.Info("trying performing history requests", zap.Uint("try", tries), zap.Stringer("peerID", peerID))
+
+		// Peform request
+		err := fn()
+		if err == nil {
+			// Reset failed requests
+			m.logger.Debug("history request performed successfully", zap.Stringer("peerID", peerID))
+			m.failedRequests[peerID] = 0
+			return nil
+		}
+
+		m.logger.Error("failed to perform history request",
+			zap.Stringer("peerID", peerID),
+			zap.Uint("tries", tries),
+			zap.Error(err),
+		)
+
+		tries++
+
+		if storeErr, ok := err.(*store.StoreError); ok {
+			if storeErr.Code == http.StatusTooManyRequests {
+				m.disconnectActiveStorenode(defaultBackoff)
+				return fmt.Errorf("ratelimited at storenode %s: %w", peerID, err)
+			}
+		}
+
+		// Increment failed requests
+		m.failedRequests[peerID]++
+
+		// Change storenode
+		if m.failedRequests[peerID] >= storenodeMaxFailedRequests {
+			return errors.New("too many failed requests")
+		}
+		// Wait a couple of second not to spam
+		time.Sleep(2 * time.Second)
+
+	}
+	return errors.New("failed to perform history request")
+}
@@ -0,0 +1,48 @@
+package history
+
+import "sync"
+
+type Emitter[T any] struct {
+	sync.Mutex
+	subscriptions []chan T
+}
+
+func NewEmitter[T any]() *Emitter[T] {
+	return &Emitter[T]{}
+}
+
+func (s *Emitter[T]) Subscribe() <-chan T {
+	s.Lock()
+	defer s.Unlock()
+	c := make(chan T)
+	s.subscriptions = append(s.subscriptions, c)
+	return c
+}
+
+func (s *Emitter[T]) Emit(value T) {
+	s.Lock()
+	defer s.Unlock()
+
+	for _, sub := range s.subscriptions {
+		sub <- value
+	}
+}
+
+type OneShotEmitter[T any] struct {
+	Emitter[T]
+}
+
+func NewOneshotEmitter[T any]() *OneShotEmitter[T] {
+	return &OneShotEmitter[T]{}
+}
+
+func (s *OneShotEmitter[T]) Emit(value T) {
+	s.Lock()
+	defer s.Unlock()
+
+	for _, subs := range s.subscriptions {
+		subs <- value
+		close(subs)
+	}
+	s.subscriptions = nil
+}
@@ -0,0 +1,67 @@
+package history
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestEmitter(t *testing.T) {
+	emitter := NewEmitter[int]()
+
+	subscr1 := emitter.Subscribe()
+	subscr2 := emitter.Subscribe()
+
+	wg := sync.WaitGroup{}
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+		emitter.Emit(1)
+		emitter.Emit(2)
+	}()
+
+	go func() {
+		defer wg.Done()
+		require.Equal(t, 1, <-subscr1)
+		require.Equal(t, 2, <-subscr1)
+	}()
+
+	go func() {
+		defer wg.Done()
+		require.Equal(t, 1, <-subscr2)
+		require.Equal(t, 2, <-subscr2)
+	}()
+
+	wg.Wait()
+}
+
+func TestOneShotEmitter(t *testing.T) {
+	emitter := NewOneshotEmitter[struct{}]()
+
+	subscr1 := emitter.Subscribe()
+	subscr2 := emitter.Subscribe()
+
+	wg := sync.WaitGroup{}
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+		emitter.Emit(struct{}{})
+	}()
+
+	go func() {
+		defer wg.Done()
+		for range subscr1 {
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for range subscr2 {
+		}
+	}()
+
+	wg.Wait()
+}
@@ -0,0 +1,331 @@
+package history
+
+import (
+	"context"
+	"encoding/hex"
+	"errors"
+	"math"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/logging"
+	"github.com/waku-org/go-waku/waku/v2/api/common"
+	"github.com/waku-org/go-waku/waku/v2/protocol"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
+	"google.golang.org/protobuf/proto"
+
+	"go.uber.org/zap"
+)
+
+const maxTopicsPerRequest int = 10
+const mailserverRequestTimeout = 30 * time.Second
+
+type work struct {
+	criteria store.FilterCriteria
+	cursor   []byte
+	limit    uint64
+}
+
+type HistoryRetriever struct {
+	store            common.StorenodeRequestor
+	logger           *zap.Logger
+	historyProcessor HistoryProcessor
+}
+
+type HistoryProcessor interface {
+	OnEnvelope(env *protocol.Envelope, processEnvelopes bool) error
+	OnRequestFailed(requestID []byte, peerID peer.ID, err error)
+}
+
+func NewHistoryRetriever(store common.StorenodeRequestor, historyProcessor HistoryProcessor, logger *zap.Logger) *HistoryRetriever {
+	return &HistoryRetriever{
+		store:            store,
+		logger:           logger.Named("history-retriever"),
+		historyProcessor: historyProcessor,
+	}
+}
+
+func (hr *HistoryRetriever) Query(
+	ctx context.Context,
+	criteria store.FilterCriteria,
+	storenodeID peer.ID,
+	pageLimit uint64,
+	shouldProcessNextPage func(int) (bool, uint64),
+	processEnvelopes bool,
+) error {
+	logger := hr.logger.With(
+		logging.Timep("fromString", criteria.TimeStart),
+		logging.Timep("toString", criteria.TimeEnd),
+		zap.String("pubsubTopic", criteria.PubsubTopic),
+		zap.Strings("contentTopics", criteria.ContentTopicsList()),
+		zap.Int64p("from", criteria.TimeStart),
+		zap.Int64p("to", criteria.TimeEnd),
+	)
+
+	logger.Info("syncing")
+
+	wg := sync.WaitGroup{}
+	workWg := sync.WaitGroup{}
+	workCh := make(chan work, 1000)       // each batch item is split in 10 topics bunch and sent to this channel
+	workCompleteCh := make(chan struct{}) // once all batch items are processed, this channel is triggered
+	semaphore := make(chan struct{}, 3)   // limit the number of concurrent queries
+	errCh := make(chan error)
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// TODO: refactor this by extracting the consumer into a separate go routine.
+
+	// Producer
+	wg.Add(1)
+	go func() {
+		defer func() {
+			logger.Debug("mailserver batch producer complete")
+			wg.Done()
+		}()
+
+		contentTopicList := criteria.ContentTopics.ToList()
+
+		// TODO: split into 24h batches
+
+		allWorks := int(math.Ceil(float64(len(contentTopicList)) / float64(maxTopicsPerRequest)))
+		workWg.Add(allWorks)
+
+		for i := 0; i < len(contentTopicList); i += maxTopicsPerRequest {
+			j := i + maxTopicsPerRequest
+			if j > len(contentTopicList) {
+				j = len(contentTopicList)
+			}
+
+			select {
+			case <-ctx.Done():
+				logger.Debug("processBatch producer - context done")
+				return
+			default:
+				logger.Debug("processBatch producer - creating work")
+				workCh <- work{
+					criteria: store.FilterCriteria{
+						ContentFilter: protocol.NewContentFilter(criteria.PubsubTopic, contentTopicList[i:j]...),
+						TimeStart:     criteria.TimeStart,
+						TimeEnd:       criteria.TimeEnd,
+					},
+					limit: pageLimit,
+				}
+			}
+		}
+
+		go func() {
+			workWg.Wait()
+			workCompleteCh <- struct{}{}
+		}()
+
+		logger.Debug("processBatch producer complete")
+	}()
+
+	var result error
+
+loop:
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Debug("processBatch cleanup - context done")
+			result = ctx.Err()
+			if errors.Is(result, context.Canceled) {
+				result = nil
+			}
+			break loop
+		case w, ok := <-workCh:
+			if !ok {
+				continue
+			}
+
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			default:
+				// continue...
+			}
+
+			logger.Debug("processBatch - received work")
+
+			semaphore <- struct{}{}
+			go func(w work) { // Consumer
+				defer func() {
+					workWg.Done()
+					<-semaphore
+				}()
+
+				queryCtx, queryCancel := context.WithTimeout(ctx, mailserverRequestTimeout)
+
+				// If time range is greater than 24 hours, limit the range: to - (to-24h)
+				// TODO: handle cases in which TimeStart/TimeEnd could be nil
+				//       (this type of query does not happen in status-go, though, and
+				//       nwaku might limit query duration to 24h anyway, so perhaps
+				//       it's not worth adding such logic)
+				timeStart := w.criteria.TimeStart
+				timeEnd := w.criteria.TimeEnd
+				exceeds24h := false
+				if timeStart != nil && timeEnd != nil && *timeEnd-*timeStart > (24*time.Hour).Nanoseconds() {
+					newTimeStart := *timeEnd - (24 * time.Hour).Nanoseconds()
+					timeStart = &newTimeStart
+					exceeds24h = true
+				}
+
+				newCriteria := w.criteria
+				newCriteria.TimeStart = timeStart
+				newCriteria.TimeEnd = timeEnd
+
+				cursor, envelopesCount, err := hr.createMessagesRequest(queryCtx, storenodeID, newCriteria, w.cursor, w.limit, true, processEnvelopes, logger)
+				queryCancel()
+
+				if err != nil {
+					logger.Debug("failed to send request", zap.Error(err))
+					errCh <- err
+					return
+				}
+
+				processNextPage := true
+				nextPageLimit := pageLimit
+				if shouldProcessNextPage != nil {
+					processNextPage, nextPageLimit = shouldProcessNextPage(envelopesCount)
+				}
+
+				if !processNextPage {
+					return
+				}
+
+				// Check the cursor after calling `shouldProcessNextPage`.
+				// The app might use process the fetched envelopes in the callback for own needs.
+				// If from/to does not exceed 24h and no cursor was returned, we have already
+				// requested the entire time range
+				if cursor == nil && !exceeds24h {
+					return
+				}
+
+				logger.Debug("processBatch producer - creating work (cursor)")
+
+				newWork := work{
+					criteria: w.criteria,
+					cursor:   cursor,
+					limit:    nextPageLimit,
+				}
+
+				// If from/to has exceeded the 24h, but there are no more records within the current
+				// 24h range, then we update the `to` for the new work to not include it.
+				if cursor == nil && exceeds24h {
+					newWork.criteria.TimeEnd = timeStart
+				}
+
+				workWg.Add(1)
+				workCh <- newWork
+			}(w)
+		case err := <-errCh:
+			logger.Debug("processBatch - received error", zap.Error(err))
+			cancel() // Kill go routines
+			return err
+		case <-workCompleteCh:
+			logger.Debug("processBatch - all jobs complete")
+			cancel() // Kill go routines
+		}
+	}
+
+	wg.Wait()
+
+	logger.Info("synced topic", zap.NamedError("hasError", result))
+
+	return result
+}
+
+func (hr *HistoryRetriever) createMessagesRequest(
+	ctx context.Context,
+	peerID peer.ID,
+	criteria store.FilterCriteria,
+	cursor []byte,
+	limit uint64,
+	waitForResponse bool,
+	processEnvelopes bool,
+	logger *zap.Logger,
+) (storeCursor []byte, envelopesCount int, err error) {
+	if waitForResponse {
+		resultCh := make(chan struct {
+			storeCursor    []byte
+			envelopesCount int
+			err            error
+		})
+
+		go func() {
+			storeCursor, envelopesCount, err = hr.requestStoreMessages(ctx, peerID, criteria, cursor, limit, processEnvelopes)
+			resultCh <- struct {
+				storeCursor    []byte
+				envelopesCount int
+				err            error
+			}{storeCursor, envelopesCount, err}
+		}()
+
+		select {
+		case result := <-resultCh:
+			return result.storeCursor, result.envelopesCount, result.err
+		case <-ctx.Done():
+			return nil, 0, ctx.Err()
+		}
+	} else {
+		go func() {
+			_, _, err = hr.requestStoreMessages(ctx, peerID, criteria, cursor, limit, false)
+			if err != nil {
+				logger.Error("failed to request store messages", zap.Error(err))
+			}
+		}()
+	}
+
+	return
+}
+
+func (hr *HistoryRetriever) requestStoreMessages(ctx context.Context, peerID peer.ID, 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.Debug("store.query",
+		logging.Timep("startTime", criteria.TimeStart),
+		logging.Timep("endTime", criteria.TimeEnd),
+		zap.Strings("contentTopics", criteria.ContentTopics.ToList()),
+		zap.String("pubsubTopic", criteria.PubsubTopic),
+		zap.String("cursor", hexutil.Encode(cursor)),
+	)
+
+	storeQueryRequest := &pb.StoreQueryRequest{
+		RequestId:        hex.EncodeToString(requestID),
+		IncludeData:      true,
+		PubsubTopic:      &criteria.PubsubTopic,
+		ContentTopics:    criteria.ContentTopicsList(),
+		TimeStart:        criteria.TimeStart,
+		TimeEnd:          criteria.TimeEnd,
+		PaginationCursor: cursor,
+		PaginationLimit:  proto.Uint64(limit),
+	}
+
+	queryStart := time.Now()
+	result, err := hr.store.Query(ctx, peerID, storeQueryRequest)
+	queryDuration := time.Since(queryStart)
+	if err != nil {
+		logger.Error("error querying storenode", zap.Error(err))
+
+		hr.historyProcessor.OnRequestFailed(requestID, peerID, err)
+
+		return nil, 0, err
+	}
+
+	messages := result.Messages()
+	envelopesCount := len(messages)
+	logger.Debug("store.query response", zap.Duration("queryDuration", queryDuration), zap.Int("numMessages", envelopesCount), zap.Bool("hasCursor", result.IsComplete() && result.Cursor() != nil))
+	for _, mkv := range messages {
+		envelope := protocol.NewEnvelope(mkv.Message, mkv.Message.GetTimestamp(), mkv.GetPubsubTopic())
+		err := hr.historyProcessor.OnEnvelope(envelope, processEnvelopes)
+		if err != nil {
+			return nil, 0, err
+		}
+	}
+	return result.Cursor(), envelopesCount, nil
+}
@@ -0,0 +1,251 @@
+package history
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"math/big"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+	"github.com/waku-org/go-waku/waku/v2/api/common"
+	"github.com/waku-org/go-waku/waku/v2/protocol"
+	proto_pb "github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/waku-org/go-waku/waku/v2/utils"
+)
+
+type queryResponse struct {
+	contentTopics []string
+	messages      []*pb.WakuMessageKeyValue
+	err           error // Indicates if this response will simulate an error returned by SendMessagesRequestForTopics
+	cursor        []byte
+}
+
+type mockResult struct {
+	cursor   []byte
+	messages []*pb.WakuMessageKeyValue
+}
+
+func (r *mockResult) Cursor() []byte {
+	return r.cursor
+}
+
+func (r *mockResult) Messages() []*pb.WakuMessageKeyValue {
+	return r.messages
+}
+
+func (r *mockResult) IsComplete() bool {
+	return false
+}
+
+func (r *mockResult) PeerID() peer.ID {
+	return ""
+}
+
+func (r *mockResult) Query() *pb.StoreQueryRequest {
+	return nil
+}
+
+func (r *mockResult) Response() *pb.StoreQueryResponse {
+	return nil
+}
+
+func (r *mockResult) Next(ctx context.Context, opts ...store.RequestOption) error {
+	return nil
+}
+
+type mockHistoryProcessor struct {
+}
+
+func (h *mockHistoryProcessor) OnEnvelope(env *protocol.Envelope, processEnvelopes bool) error {
+	return nil
+}
+
+func (h *mockHistoryProcessor) OnRequestFailed(requestID []byte, peerID peer.ID, err error) {
+}
+
+func newMockHistoryProcessor() *mockHistoryProcessor {
+	return &mockHistoryProcessor{}
+}
+
+type mockStore struct {
+	queryResponses map[string]queryResponse
+}
+
+func newMockStore() *mockStore {
+	return &mockStore{
+		queryResponses: make(map[string]queryResponse),
+	}
+}
+
+func getInitialResponseKey(contentTopics []string) string {
+	sort.Strings(contentTopics)
+	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) {
+	result := &mockResult{}
+	if len(storeQueryRequest.GetPaginationCursor()) == 0 {
+		initialResponse := getInitialResponseKey(storeQueryRequest.GetContentTopics())
+		response := t.queryResponses[initialResponse]
+		if response.err != nil {
+			return nil, response.err
+		}
+		result.cursor = response.cursor
+		result.messages = response.messages
+	} else {
+		response := t.queryResponses[hex.EncodeToString(storeQueryRequest.GetPaginationCursor())]
+		if response.err != nil {
+			return nil, response.err
+		}
+		result.cursor = response.cursor
+		result.messages = response.messages
+	}
+
+	return result, nil
+}
+
+func (t *mockStore) Populate(topics []string, responses int, includeRandomError bool) error {
+	if responses <= 0 || len(topics) == 0 {
+		return errors.New("invalid input parameters")
+	}
+
+	var topicBatches [][]string
+
+	for i := 0; i < len(topics); i += maxTopicsPerRequest {
+		// Split batch in 10-contentTopic subbatches
+		j := i + maxTopicsPerRequest
+		if j > len(topics) {
+			j = len(topics)
+		}
+		topicBatches = append(topicBatches, topics[i:j])
+	}
+
+	randomErrIdx, err := rand.Int(rand.Reader, big.NewInt(int64(len(topicBatches))))
+	if err != nil {
+		return err
+	}
+	randomErrIdxInt := int(randomErrIdx.Int64())
+
+	for i, topicBatch := range topicBatches {
+		// Setup initial response
+		initialResponseKey := getInitialResponseKey(topicBatch)
+		t.queryResponses[initialResponseKey] = queryResponse{
+			contentTopics: topicBatch,
+			messages: []*pb.WakuMessageKeyValue{
+				{
+					MessageHash: protocol.GenerateRequestID(),
+					Message: &proto_pb.WakuMessage{
+						Payload:      []byte{1, 2, 3},
+						ContentTopic: "abc",
+						Timestamp:    proto.Int64(time.Now().UnixNano()),
+					},
+					PubsubTopic: proto.String("test"),
+				},
+			},
+			err: nil,
+		}
+
+		prevKey := initialResponseKey
+		for x := 0; x < responses-1; x++ {
+			newResponseCursor := []byte(uuid.New().String())
+			newResponseKey := hex.EncodeToString(newResponseCursor)
+
+			var err error
+			if includeRandomError && i == randomErrIdxInt && x == responses-2 { // Include an error in last request
+				err = errors.New("random error")
+			}
+
+			t.queryResponses[newResponseKey] = queryResponse{
+				contentTopics: topicBatch,
+				messages: []*pb.WakuMessageKeyValue{
+					{
+						MessageHash: protocol.GenerateRequestID(),
+						Message: &proto_pb.WakuMessage{
+							Payload:      []byte{1, 2, 3},
+							ContentTopic: "abc",
+							Timestamp:    proto.Int64(time.Now().UnixNano()),
+						},
+						PubsubTopic: proto.String("test"),
+					},
+				},
+				err: err,
+			}
+
+			// Updating prev response cursor to point to the new response
+			prevResponse := t.queryResponses[prevKey]
+			prevResponse.cursor = newResponseCursor
+			t.queryResponses[prevKey] = prevResponse
+
+			prevKey = newResponseKey
+		}
+
+	}
+
+	return nil
+}
+
+func TestSuccessBatchExecution(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
+	defer cancel()
+
+	storenodeID, err := peer.Decode("16Uiu2HAkw3x97MbbZSWHbdF5bob45vcZvPPK4s4Mjyv2mxyB9GS3")
+	require.NoError(t, err)
+
+	topics := []string{}
+	for i := 0; i < 50; i++ {
+		topics = append(topics, uuid.NewString())
+	}
+
+	testStore := newMockStore()
+	err = testStore.Populate(topics, 10, false)
+	require.NoError(t, err)
+
+	historyProcessor := newMockHistoryProcessor()
+
+	historyRetriever := NewHistoryRetriever(testStore, historyProcessor, utils.Logger())
+
+	criteria := store.FilterCriteria{
+		ContentFilter: protocol.NewContentFilter("test", topics...),
+	}
+
+	err = historyRetriever.Query(ctx, criteria, storenodeID, 10, func(i int) (bool, uint64) { return true, 10 }, true)
+	require.NoError(t, err)
+}
+
+func TestFailedBatchExecution(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
+	defer cancel()
+
+	storenodeID, err := peer.Decode("16Uiu2HAkw3x97MbbZSWHbdF5bob45vcZvPPK4s4Mjyv2mxyB9GS3")
+	require.NoError(t, err)
+
+	topics := []string{}
+	for i := 0; i < 2; i++ {
+		topics = append(topics, uuid.NewString())
+	}
+
+	testStore := newMockStore()
+	err = testStore.Populate(topics, 10, true)
+	require.NoError(t, err)
+
+	historyProcessor := newMockHistoryProcessor()
+
+	historyRetriever := NewHistoryRetriever(testStore, historyProcessor, utils.Logger())
+
+	criteria := store.FilterCriteria{
+		ContentFilter: protocol.NewContentFilter("test", topics...),
+	}
+
+	err = historyRetriever.Query(ctx, criteria, storenodeID, 10, func(i int) (bool, uint64) { return true, 10 }, true)
+	require.Error(t, err)
+}
@@ -0,0 +1,32 @@
+package history
+
+import (
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+type SortedStorenode struct {
+	Storenode       peer.AddrInfo
+	RTT             time.Duration
+	CanConnectAfter time.Time
+}
+
+type byRTTMsAndCanConnectBefore []SortedStorenode
+
+func (s byRTTMsAndCanConnectBefore) Len() int {
+	return len(s)
+}
+
+func (s byRTTMsAndCanConnectBefore) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+
+func (s byRTTMsAndCanConnectBefore) Less(i, j int) bool {
+	// Slightly inaccurate as time sensitive sorting, but it does not matter so much
+	now := time.Now()
+	if s[i].CanConnectAfter.Before(now) && s[j].CanConnectAfter.Before(now) {
+		return s[i].RTT < s[j].RTT
+	}
+	return s[i].CanConnectAfter.Before(s[j].CanConnectAfter)
+}
@@ -0,0 +1,29 @@
+package missing
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/api/common"
+	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+	storepb "github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
+)
+
+func NewDefaultStorenodeRequestor(store *store.WakuStore) common.StorenodeRequestor {
+	return &defaultStorenodeRequestor{
+		store: store,
+	}
+}
+
+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) Query(ctx context.Context, peerID peer.ID, storeQueryRequest *storepb.StoreQueryRequest) (common.StoreRequestResult, error) {
+	return d.store.RequestRaw(ctx, peerID, storeQueryRequest)
+}
@@ -11,9 +11,10 @@ import (

 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/waku-org/go-waku/logging"
+	"github.com/waku-org/go-waku/waku/v2/api/common"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
-	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+	storepb "github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
 	"github.com/waku-org/go-waku/waku/v2/timesource"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
@@ -22,6 +23,7 @@ import (

 const maxContentTopicsPerRequest = 10
 const maxMsgHashesPerRequest = 50
+const messageFetchPageSize = 100

 // MessageTracker should keep track of messages it has seen before and
 // provide a way to determine whether a message exists or not. This
@@ -33,22 +35,25 @@ 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

-	messageTracker MessageTracker
+	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

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

 // NewMissingMessageVerifier creates an instance of a MissingMessageVerifier
-func NewMissingMessageVerifier(store *store.WakuStore, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier {
+func NewMissingMessageVerifier(storenodeRequester common.StorenodeRequestor, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier {
 	options = append(defaultMissingMessagesVerifierOptions, options...)
 	params := missingMessageVerifierParams{}
 	for _, opt := range options {
@@ -56,11 +61,13 @@ func NewMissingMessageVerifier(store *store.WakuStore, messageTracker MessageTra
 	}

 	return &MissingMessageVerifier{
-		store:          store,
-		timesource:     timesource,
-		messageTracker: messageTracker,
-		logger:         logger.Named("missing-msg-verifier"),
-		params:         params,
+		storenodeRequestor: storenodeRequester,
+		timesource:         timesource,
+		messageTracker:     messageTracker,
+		logger:             logger.Named("missing-msg-verifier"),
+		params:             params,
+		criteriaInterest:   make(map[string]*criteriaInterest),
+		C:                  make(chan *protocol.Envelope, 1000),
 	}
 }

@@ -92,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()
@@ -115,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 {
@@ -178,7 +215,7 @@ func (m *MissingMessageVerifier) fetchHistory(c chan<- *protocol.Envelope, inter
 	}
 }

-func (m *MissingMessageVerifier) storeQueryWithRetry(ctx context.Context, queryFunc func(ctx context.Context) (*store.Result, error), logger *zap.Logger, logMsg string) (*store.Result, error) {
+func (m *MissingMessageVerifier) storeQueryWithRetry(ctx context.Context, queryFunc func(ctx context.Context) (common.StoreRequestResult, error), logger *zap.Logger, logMsg string) (common.StoreRequestResult, error) {
 	retry := true
 	count := 1
 	for retry && count <= m.params.maxAttemptsToRetrieveHistory {
@@ -212,12 +249,21 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 		logging.Epoch("to", now),
 	)

-	result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (*store.Result, error) {
-		return m.store.Query(ctx, store.FilterCriteria{
-			ContentFilter: protocol.NewContentFilter(interest.contentFilter.PubsubTopic, contentTopics[batchFrom:batchTo]...),
-			TimeStart:     proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()),
-			TimeEnd:       proto.Int64(now.Add(-m.params.delay).UnixNano()),
-		}, store.WithPeer(interest.peerID), store.WithPaging(false, 100), store.IncludeData(false))
+	result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
+		storeQueryRequest := &storepb.StoreQueryRequest{
+			RequestId:       hex.EncodeToString(protocol.GenerateRequestID()),
+			PubsubTopic:     &interest.contentFilter.PubsubTopic,
+			ContentTopics:   contentTopics[batchFrom:batchTo],
+			TimeStart:       proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()),
+			TimeEnd:         proto.Int64(now.Add(-m.params.delay).UnixNano()),
+			PaginationLimit: proto.Uint64(messageFetchPageSize),
+		}
+
+		return m.storenodeRequestor.Query(
+			ctx,
+			interest.peerID,
+			storeQueryRequest,
+		)
 	}, logger, "retrieving history to check for missing messages")
 	if err != nil {
 		if !errors.Is(err, context.Canceled) {
@@ -243,7 +289,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 			missingHashes = append(missingHashes, hash)
 		}

-		result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (*store.Result, error) {
+		result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
 			if err = result.Next(ctx); err != nil {
 				return nil, err
 			}
@@ -282,10 +328,23 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 			defer utils.LogOnPanic()
 			defer wg.Wait()

-			result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (*store.Result, error) {
+			result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
 				queryCtx, cancel := context.WithTimeout(ctx, m.params.storeQueryTimeout)
 				defer cancel()
-				return m.store.QueryByHash(queryCtx, messageHashes, store.WithPeer(interest.peerID), store.WithPaging(false, maxMsgHashesPerRequest))
+
+				var messageHashesBytes [][]byte
+				for _, m := range messageHashes {
+					messageHashesBytes = append(messageHashesBytes, m.Bytes())
+				}
+
+				storeQueryRequest := &storepb.StoreQueryRequest{
+					RequestId:       hex.EncodeToString(protocol.GenerateRequestID()),
+					IncludeData:     true,
+					MessageHashes:   messageHashesBytes,
+					PaginationLimit: proto.Uint64(maxMsgHashesPerRequest),
+				}
+
+				return m.storenodeRequestor.Query(queryCtx, interest.peerID, storeQueryRequest)
 			}, logger, "retrieving missing messages")
 			if err != nil {
 				if !errors.Is(err, context.Canceled) {
@@ -303,7 +362,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 					}
 				}

-				result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (*store.Result, error) {
+				result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
 					if err = result.Next(ctx); err != nil {
 						return nil, err
 					}
@@ -0,0 +1,50 @@
+package publish
+
+import (
+	"context"
+	"errors"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/protocol/lightpush"
+	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
+)
+
+var ErrRelayNotAvailable = errors.New("relay is not available")
+var ErrLightpushNotAvailable = errors.New("lightpush is not available")
+
+func NewDefaultPublisher(lightpush *lightpush.WakuLightPush, relay *relay.WakuRelay) Publisher {
+	return &defaultPublisher{
+		lightpush: lightpush,
+		relay:     relay,
+	}
+}
+
+type defaultPublisher struct {
+	lightpush *lightpush.WakuLightPush
+	relay     *relay.WakuRelay
+}
+
+func (d *defaultPublisher) RelayListPeers(pubsubTopic string) ([]peer.ID, error) {
+	if d.relay == nil {
+		return nil, ErrRelayNotAvailable
+	}
+
+	return d.relay.PubSub().ListPeers(pubsubTopic), nil
+}
+
+func (d *defaultPublisher) RelayPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string) (pb.MessageHash, error) {
+	if d.relay == nil {
+		return pb.MessageHash{}, ErrRelayNotAvailable
+	}
+
+	return d.relay.Publish(ctx, message, relay.WithPubSubTopic(pubsubTopic))
+}
+
+func (d *defaultPublisher) LightpushPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string, maxPeers int) (pb.MessageHash, error) {
+	if d.lightpush == nil {
+		return pb.MessageHash{}, ErrLightpushNotAvailable
+	}
+
+	return d.lightpush.Publish(ctx, message, lightpush.WithPubSubTopic(pubsubTopic), lightpush.WithMaxPeers(maxPeers))
+}
@@ -0,0 +1,39 @@
+package publish
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+)
+
+func NewDefaultStorenodeMessageVerifier(store *store.WakuStore) StorenodeMessageVerifier {
+	return &defaultStorenodeMessageVerifier{
+		store: store,
+	}
+}
+
+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) {
+	var opts []store.RequestOption
+	opts = append(opts, store.WithRequestID(requestID))
+	opts = append(opts, store.WithPeer(peerID))
+	opts = append(opts, store.WithPaging(false, pageSize))
+	opts = append(opts, store.IncludeData(false))
+
+	response, err := d.store.QueryByHash(ctx, messageHashes, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	result := make([]pb.MessageHash, len(response.Messages()))
+	for i, msg := range response.Messages() {
+		result[i] = msg.WakuMessageHash()
+	}
+
+	return result, nil
+}
@@ -10,9 +10,9 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/libp2p/go-libp2p/core/peer"
 	apicommon "github.com/waku-org/go-waku/waku/v2/api/common"
+	"github.com/waku-org/go-waku/waku/v2/api/history"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
-	"github.com/waku-org/go-waku/waku/v2/protocol/store"
 	"github.com/waku-org/go-waku/waku/v2/timesource"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
@@ -29,7 +29,11 @@ type ISentCheck interface {
 	Start()
 	Add(topic string, messageID common.Hash, sentTime uint32)
 	DeleteByMessageIDs(messageIDs []common.Hash)
-	SetStorePeerID(peerID peer.ID)
+}
+
+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)
 }

 // MessageSentCheck tracks the outgoing messages and check against store node
@@ -38,11 +42,11 @@ type ISentCheck interface {
 type MessageSentCheck struct {
 	messageIDs          map[string]map[common.Hash]uint32
 	messageIDsMu        sync.RWMutex
-	storePeerID         peer.ID
 	messageStoredChan   chan common.Hash
 	messageExpiredChan  chan common.Hash
 	ctx                 context.Context
-	store               *store.WakuStore
+	messageVerifier     StorenodeMessageVerifier
+	storenodeCycle      *history.StorenodeCycle
 	timesource          timesource.Timesource
 	logger              *zap.Logger
 	maxHashQueryLength  uint64
@@ -53,14 +57,15 @@ type MessageSentCheck struct {
 }

 // NewMessageSentCheck creates a new instance of MessageSentCheck with default parameters
-func NewMessageSentCheck(ctx context.Context, store *store.WakuStore, timesource timesource.Timesource, msgStoredChan chan common.Hash, msgExpiredChan chan common.Hash, logger *zap.Logger) *MessageSentCheck {
+func NewMessageSentCheck(ctx context.Context, messageVerifier StorenodeMessageVerifier, cycle *history.StorenodeCycle, timesource timesource.Timesource, msgStoredChan chan common.Hash, msgExpiredChan chan common.Hash, logger *zap.Logger) *MessageSentCheck {
 	return &MessageSentCheck{
 		messageIDs:          make(map[string]map[common.Hash]uint32),
 		messageIDsMu:        sync.RWMutex{},
 		messageStoredChan:   msgStoredChan,
 		messageExpiredChan:  msgExpiredChan,
 		ctx:                 ctx,
-		store:               store,
+		messageVerifier:     messageVerifier,
+		storenodeCycle:      cycle,
 		timesource:          timesource,
 		logger:              logger,
 		maxHashQueryLength:  DefaultMaxHashQueryLength,
@@ -139,11 +144,6 @@ func (m *MessageSentCheck) DeleteByMessageIDs(messageIDs []common.Hash) {
 	}
 }

-// SetStorePeerID sets the peer id of store node
-func (m *MessageSentCheck) SetStorePeerID(peerID peer.ID) {
-	m.storePeerID = peerID
-}
-
 // Start checks if the tracked outgoing messages are stored periodically
 func (m *MessageSentCheck) Start() {
 	defer utils.LogOnPanic()
@@ -211,18 +211,13 @@ func (m *MessageSentCheck) Start() {
 }

 func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []common.Hash, relayTime []uint32, pubsubTopic string) []common.Hash {
-	selectedPeer := m.storePeerID
+	selectedPeer := m.storenodeCycle.GetActiveStorenode()
 	if selectedPeer == "" {
 		m.logger.Error("no store peer id available", zap.String("pubsubTopic", pubsubTopic))
 		return []common.Hash{}
 	}

-	var opts []store.RequestOption
 	requestID := protocol.GenerateRequestID()
-	opts = append(opts, store.WithRequestID(requestID))
-	opts = append(opts, store.WithPeer(selectedPeer))
-	opts = append(opts, store.WithPaging(false, m.maxHashQueryLength))
-	opts = append(opts, store.IncludeData(false))

 	messageHashes := make([]pb.MessageHash, len(hashes))
 	for i, hash := range hashes {
@@ -233,20 +228,20 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c

 	queryCtx, cancel := context.WithTimeout(ctx, m.storeQueryTimeout)
 	defer cancel()
-	result, err := m.store.QueryByHash(queryCtx, messageHashes, opts...)
+	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))
 		return []common.Hash{}
 	}

-	m.logger.Debug("store.queryByHash result", zap.String("requestID", hexutil.Encode(requestID)), zap.Int("messages", len(result.Messages())))
+	m.logger.Debug("store.queryByHash result", zap.String("requestID", hexutil.Encode(requestID)), zap.Int("messages", len(result)))

 	var ackHashes []common.Hash
 	var missedHashes []common.Hash
 	for i, hash := range hashes {
 		found := false
-		for _, msg := range result.Messages() {
-			if bytes.Equal(msg.GetMessageHash(), hash.Bytes()) {
+		for _, msgHash := range result {
+			if bytes.Equal(msgHash.Bytes(), hash.Bytes()) {
 				found = true
 				break
 			}
@@ -10,7 +10,7 @@ import (

 func TestAddAndDelete(t *testing.T) {
 	ctx := context.TODO()
-	messageSentCheck := NewMessageSentCheck(ctx, nil, nil, nil, nil, nil)
+	messageSentCheck := NewMessageSentCheck(ctx, nil, nil, nil, nil, nil, nil)

 	messageSentCheck.Add("topic", [32]byte{1}, 1)
 	messageSentCheck.Add("topic", [32]byte{2}, 2)
@@ -6,17 +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/lightpush"
-	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
+	"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

@@ -37,13 +35,29 @@ func (pm PublishMethod) String() string {
 	}
 }

+type Publisher interface {
+	// RelayListPeers returns the list of peers for a pubsub topic
+	RelayListPeers(pubsubTopic string) ([]peer.ID, error)
+
+	// RelayPublish publishes a message via WakuRelay
+	RelayPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string) (pb.MessageHash, error)
+
+	// LightpushPublish publishes a message via WakuLightPush
+	LightpushPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string, maxPeers int) (pb.MessageHash, error)
+}
+
 type MessageSender struct {
 	publishMethod    PublishMethod
-	lightPush        *lightpush.WakuLightPush
-	relay            *relay.WakuRelay
+	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 {
@@ -65,15 +79,19 @@ func (r *Request) WithPublishMethod(publishMethod PublishMethod) *Request {
 	return r
 }

-func NewMessageSender(publishMethod PublishMethod, lightPush *lightpush.WakuLightPush, relay *relay.WakuRelay, 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,
-		lightPush:     lightPush,
-		relay:         relay,
-		rateLimiter:   NewPublishRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst),
+		publisher:     publisher,
+		rateLimiter:   rateLimiter,
 		logger:        logger,
 	}, nil
 }
@@ -83,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()),
@@ -109,26 +136,23 @@ func (ms *MessageSender) Send(req *Request) error {

 	switch publishMethod {
 	case LightPush:
-		if ms.lightPush == nil {
-			return errors.New("lightpush is not available")
-		}
 		logger.Info("publishing message via lightpush")
-		_, err := ms.lightPush.Publish(
+		_, err := ms.publisher.LightpushPublish(
 			req.ctx,
 			req.envelope.Message(),
-			lightpush.WithPubSubTopic(req.envelope.PubsubTopic()),
-			lightpush.WithMaxPeers(DefaultPeersToPublishForLightpush),
+			req.envelope.PubsubTopic(),
+			DefaultPeersToPublishForLightpush,
 		)
 		if err != nil {
 			return err
 		}
 	case Relay:
-		if ms.relay == nil {
-			return errors.New("relay is not available")
+		peers, err := ms.publisher.RelayListPeers(req.envelope.PubsubTopic())
+		if err != nil {
+			return err
 		}
-		peerCnt := len(ms.relay.PubSub().ListPeers(req.envelope.PubsubTopic()))
-		logger.Info("publishing message via relay", zap.Int("peerCnt", peerCnt))
-		_, err := ms.relay.Publish(req.ctx, req.envelope.Message(), relay.WithPubSubTopic(req.envelope.PubsubTopic()))
+		logger.Info("publishing message via relay", zap.Int("peerCnt", len(peers)))
+		_, err = ms.publisher.RelayPublish(req.ctx, req.envelope.Message(), req.envelope.PubsubTopic())
 		if err != nil {
 			return err
 		}
@@ -144,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
 }

@@ -162,9 +196,3 @@ func (ms *MessageSender) MessagesDelivered(messageIDs []common.Hash) {
 		ms.messageSentCheck.DeleteByMessageIDs(messageIDs)
 	}
 }
-
-func (ms *MessageSender) SetStorePeerID(peerID peer.ID) {
-	if ms.messageSentCheck != nil {
-		ms.messageSentCheck.SetStorePeerID(peerID)
-	}
-}
@@ -3,6 +3,7 @@ package publish
 import (
 	"context"
 	"crypto/rand"
+	"sync"
 	"testing"
 	"time"

@@ -53,7 +54,8 @@ func TestNewSenderWithRelay(t *testing.T) {

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

 	_, err = relayNode.Subscribe(context.Background(), protocol.NewContentFilter("test-pubsub-topic"))
 	require.Nil(t, err)
-	sender, err := NewMessageSender(Relay, nil, relayNode, utils.Logger())
+	publisher := NewDefaultPublisher(nil, relayNode)
+	sender, err := NewMessageSender(Relay, publisher, nil, utils.Logger())

 	check := &MockMessageSentCheck{Messages: make(map[string]map[common.Hash]uint32)}
 	sender.WithMessageSentCheck(check)
@@ -127,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)
+}
@@ -464,13 +464,17 @@ func (w *WakuNode) Start(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		//TODO: setting this up temporarily to improve connectivity success for lightNode in status.
-		//This will have to be removed or changed with community sharding will be implemented.
-		if w.opts.shards != nil {
-			err = w.SetRelayShards(*w.opts.shards)
-			if err != nil {
-				return err
-			}
+	}
+
+	//TODO: setting this up temporarily to improve connectivity success for lightNode
+	//      in status. Also, when executing go-waku service-node as a lightclient
+	//      (using --pubsub-topic and --relay=false)
+	//      This will have to be removed or changed with community sharding will be
+	//      implemented.
+	if w.opts.shards != nil {
+		err = w.SetRelayShards(*w.opts.shards)
+		if err != nil {
+			return err
 		}
 	}

@@ -101,7 +101,8 @@ const (
 const maxFailedAttempts = 5
 const prunePeerStoreInterval = 10 * time.Minute
 const peerConnectivityLoopSecs = 15
-const maxConnsToPeerRatio = 5
+const maxConnsToPeerRatio = 3
+const maxDialFailures = 2

 // 80% relay peers 20% service peers
 func relayAndServicePeers(maxConnections int) (int, int) {
@@ -256,6 +257,17 @@ func (pm *PeerManager) Start(ctx context.Context) {
 	}
 }

+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)
+		}
+	}
+}
+
 func (pm *PeerManager) peerStoreLoop(ctx context.Context) {
 	defer utils.LogOnPanic()
 	t := time.NewTicker(prunePeerStoreInterval)
@@ -731,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)
 	}
@@ -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"
@@ -50,8 +51,8 @@ type StoreError struct {
 }

 // NewStoreError creates a new instance of StoreError
-func NewStoreError(code int, message string) StoreError {
-	return StoreError{
+func NewStoreError(code int, message string) *StoreError {
+	return &StoreError{
 		Code:    code,
 		Message: message,
 	}
@@ -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
@@ -99,7 +101,7 @@ func (s *WakuStore) SetHost(h host.Host) {
 // Request is used to send a store query. This function requires understanding how to prepare a store query
 // and most of the time you can use `Query`, `QueryByHash` and `Exists` instead, as they provide
 // a simpler API
-func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...RequestOption) (*Result, error) {
+func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...RequestOption) (Result, error) {
 	params := new(Parameters)

 	optList := DefaultOptions()
@@ -182,7 +184,36 @@ func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...Requ
 		return nil, err
 	}

-	result := &Result{
+	result := &resultImpl{
+		store:         s,
+		messages:      response.Messages,
+		storeRequest:  storeRequest,
+		storeResponse: response,
+		peerID:        params.selectedPeer,
+		cursor:        response.PaginationCursor,
+	}
+
+	return result, nil
+}
+
+func (s *WakuStore) RequestRaw(ctx context.Context, peerID peer.ID, storeRequest *pb.StoreQueryRequest) (Result, error) {
+	err := storeRequest.Validate()
+	if err != nil {
+		return nil, err
+	}
+
+	var params Parameters
+	params.selectedPeer = peerID
+	if params.selectedPeer == "" {
+		return nil, ErrMustSelectPeer
+	}
+
+	response, err := s.queryFrom(ctx, storeRequest, &params)
+	if err != nil {
+		return nil, err
+	}
+
+	result := &resultImpl{
 		store:         s,
 		messages:      response.Messages,
 		storeRequest:  storeRequest,
@@ -195,12 +226,12 @@ func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...Requ
 }

 // Query retrieves all the messages that match a criteria. Use the options to indicate whether to return the message themselves or not.
-func (s *WakuStore) Query(ctx context.Context, criteria FilterCriteria, opts ...RequestOption) (*Result, error) {
+func (s *WakuStore) Query(ctx context.Context, criteria FilterCriteria, opts ...RequestOption) (Result, error) {
 	return s.Request(ctx, criteria, opts...)
 }

 // Query retrieves all the messages with specific message hashes
-func (s *WakuStore) QueryByHash(ctx context.Context, messageHashes []wpb.MessageHash, opts ...RequestOption) (*Result, error) {
+func (s *WakuStore) QueryByHash(ctx context.Context, messageHashes []wpb.MessageHash, opts ...RequestOption) (Result, error) {
 	return s.Request(ctx, MessageHashCriteria{messageHashes}, opts...)
 }

@@ -214,17 +245,17 @@ func (s *WakuStore) Exists(ctx context.Context, messageHash wpb.MessageHash, opt
 		return false, err
 	}

-	return len(result.messages) != 0, nil
+	return len(result.Messages()) != 0, nil
 }

-func (s *WakuStore) next(ctx context.Context, r *Result, opts ...RequestOption) (*Result, error) {
+func (s *WakuStore) next(ctx context.Context, r Result, opts ...RequestOption) (*resultImpl, error) {
 	if r.IsComplete() {
-		return &Result{
+		return &resultImpl{
 			store:         s,
 			messages:      nil,
 			cursor:        nil,
-			storeRequest:  r.storeRequest,
-			storeResponse: r.storeResponse,
+			storeRequest:  r.Query(),
+			storeResponse: r.Response(),
 			peerID:        r.PeerID(),
 		}, nil
 	}
@@ -240,7 +271,7 @@ func (s *WakuStore) next(ctx context.Context, r *Result, opts ...RequestOption)
 		}
 	}

-	storeRequest := proto.Clone(r.storeRequest).(*pb.StoreQueryRequest)
+	storeRequest := proto.Clone(r.Query()).(*pb.StoreQueryRequest)
 	storeRequest.RequestId = hex.EncodeToString(protocol.GenerateRequestID())
 	storeRequest.PaginationCursor = r.Cursor()

@@ -249,7 +280,7 @@ func (s *WakuStore) next(ctx context.Context, r *Result, opts ...RequestOption)
 		return nil, err
 	}

-	result := &Result{
+	result := &resultImpl{
 		store:         s,
 		messages:      response.Messages,
 		storeRequest:  storeRequest,
@@ -263,16 +294,18 @@ func (s *WakuStore) next(ctx context.Context, r *Result, opts ...RequestOption)
 }

 func (s *WakuStore) queryFrom(ctx context.Context, storeRequest *pb.StoreQueryRequest, params *Parameters) (*pb.StoreQueryResponse, error) {
-	logger := s.log.With(logging.HostID("peer", params.selectedPeer), zap.String("requestId", hex.EncodeToString([]byte(storeRequest.RequestId))))
+	logger := s.log.With(logging.HostID("peer", params.selectedPeer), zap.String("requestId", storeRequest.RequestId))

 	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
@@ -317,7 +350,7 @@ func (s *WakuStore) queryFrom(ctx context.Context, storeRequest *pb.StoreQueryRe

 	if storeResponse.GetStatusCode() != ok {
 		err := NewStoreError(int(storeResponse.GetStatusCode()), storeResponse.GetStatusDesc())
-		return nil, &err
+		return nil, err
 	}
 	return storeResponse, nil
 }
@@ -128,33 +128,33 @@ func TestStoreClient(t *testing.T) {

 	// -- First page:
 	require.False(t, response.IsComplete())
-	require.Len(t, response.messages, 2)
-	require.Equal(t, response.messages[0].Message.GetTimestamp(), messages[0].GetTimestamp())
-	require.Equal(t, response.messages[1].Message.GetTimestamp(), messages[1].GetTimestamp())
+	require.Len(t, response.Messages(), 2)
+	require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[0].GetTimestamp())
+	require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[1].GetTimestamp())

 	err = response.Next(ctx)
 	require.NoError(t, err)

 	// -- Second page:
 	require.False(t, response.IsComplete())
-	require.Len(t, response.messages, 2)
-	require.Equal(t, response.messages[0].Message.GetTimestamp(), messages[2].GetTimestamp())
-	require.Equal(t, response.messages[1].Message.GetTimestamp(), messages[3].GetTimestamp())
+	require.Len(t, response.Messages(), 2)
+	require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[2].GetTimestamp())
+	require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[3].GetTimestamp())

 	err = response.Next(ctx)
 	require.NoError(t, err)

 	// -- Third page:
 	require.False(t, response.IsComplete())
-	require.Len(t, response.messages, 1)
-	require.Equal(t, response.messages[0].Message.GetTimestamp(), messages[4].GetTimestamp())
+	require.Len(t, response.Messages(), 1)
+	require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[4].GetTimestamp())

 	err = response.Next(ctx)
 	require.NoError(t, err)

 	// -- Trying to continue a completed cursor
 	require.True(t, response.IsComplete())
-	require.Len(t, response.messages, 0)
+	require.Len(t, response.Messages(), 0)

 	err = response.Next(ctx)
 	require.NoError(t, err)
@@ -165,26 +165,26 @@ func TestStoreClient(t *testing.T) {

 	// -- First page:
 	require.False(t, response.IsComplete())
-	require.Len(t, response.messages, 2)
-	require.Equal(t, response.messages[0].Message.GetTimestamp(), messages[3].GetTimestamp())
-	require.Equal(t, response.messages[1].Message.GetTimestamp(), messages[4].GetTimestamp())
+	require.Len(t, response.Messages(), 2)
+	require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[3].GetTimestamp())
+	require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[4].GetTimestamp())

 	err = response.Next(ctx)
 	require.NoError(t, err)

 	// -- Second page:
 	require.False(t, response.IsComplete())
-	require.Len(t, response.messages, 2)
-	require.Equal(t, response.messages[0].Message.GetTimestamp(), messages[1].GetTimestamp())
-	require.Equal(t, response.messages[1].Message.GetTimestamp(), messages[2].GetTimestamp())
+	require.Len(t, response.Messages(), 2)
+	require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[1].GetTimestamp())
+	require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[2].GetTimestamp())

 	err = response.Next(ctx)
 	require.NoError(t, err)

 	// -- Third page:
 	require.False(t, response.IsComplete())
-	require.Len(t, response.messages, 1)
-	require.Equal(t, response.messages[0].Message.GetTimestamp(), messages[0].GetTimestamp())
+	require.Len(t, response.Messages(), 1)
+	require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[0].GetTimestamp())

 	err = response.Next(ctx)
 	require.NoError(t, err)
@@ -197,13 +197,13 @@ func TestStoreClient(t *testing.T) {
 	// No cursor should be returned if there are no messages that match the criteria
 	response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "no-messages"), TimeStart: startTime, TimeEnd: endTime}, WithPaging(true, 2))
 	require.NoError(t, err)
-	require.Len(t, response.messages, 0)
+	require.Len(t, response.Messages(), 0)
 	require.Empty(t, response.Cursor())

 	// If the page size is larger than the number of existing messages, it should not return a cursor
 	response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: endTime}, WithPaging(true, 100))
 	require.NoError(t, err)
-	require.Len(t, response.messages, 5)
+	require.Len(t, response.Messages(), 5)
 	require.Empty(t, response.Cursor())

 	// Invalid cursors should fail
@@ -225,17 +225,17 @@ func TestStoreClient(t *testing.T) {
 	// Handle temporal history query with a zero-size time window
 	response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: startTime})
 	require.NoError(t, err)
-	require.Len(t, response.messages, 0)
+	require.Len(t, response.Messages(), 0)
 	require.Empty(t, response.Cursor())

 	// Should not include data
 	response, err = wakuStore.Request(ctx, MessageHashCriteria{MessageHashes: []pb.MessageHash{messages[0].Hash(pubsubTopic)}}, IncludeData(false), WithPeer(storenode.ID))
 	require.NoError(t, err)
-	require.Len(t, response.messages, 1)
-	require.Nil(t, response.messages[0].Message)
+	require.Len(t, response.Messages(), 1)
+	require.Nil(t, response.Messages()[0].Message)

 	response, err = wakuStore.Request(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test")}, IncludeData(false))
 	require.NoError(t, err)
-	require.GreaterOrEqual(t, len(response.messages), 1)
-	require.Nil(t, response.messages[0].Message)
+	require.GreaterOrEqual(t, len(response.Messages()), 1)
+	require.Nil(t, response.Messages()[0].Message)
 }
@@ -22,6 +22,10 @@ type Parameters struct {
 	skipRatelimit     bool
 }

+func (p *Parameters) Cursor() []byte {
+	return p.cursor
+}
+
 type RequestOption func(*Parameters) error

 // WithPeer is an option used to specify the peerID to request the message history.
@@ -8,7 +8,17 @@ import (
 )

 // Result represents a valid response from a store node
-type Result struct {
+type Result interface {
+	Cursor() []byte
+	IsComplete() bool
+	PeerID() peer.ID
+	Query() *pb.StoreQueryRequest
+	Response() *pb.StoreQueryResponse
+	Next(ctx context.Context, opts ...RequestOption) error
+	Messages() []*pb.WakuMessageKeyValue
+}
+
+type resultImpl struct {
 	done bool

 	messages      []*pb.WakuMessageKeyValue
@@ -19,27 +29,27 @@ type Result struct {
 	peerID        peer.ID
 }

-func (r *Result) Cursor() []byte {
+func (r *resultImpl) Cursor() []byte {
 	return r.cursor
 }

-func (r *Result) IsComplete() bool {
+func (r *resultImpl) IsComplete() bool {
 	return r.done
 }

-func (r *Result) PeerID() peer.ID {
+func (r *resultImpl) PeerID() peer.ID {
 	return r.peerID
 }

-func (r *Result) Query() *pb.StoreQueryRequest {
+func (r *resultImpl) Query() *pb.StoreQueryRequest {
 	return r.storeRequest
 }

-func (r *Result) Response() *pb.StoreQueryResponse {
+func (r *resultImpl) Response() *pb.StoreQueryResponse {
 	return r.storeResponse
 }

-func (r *Result) Next(ctx context.Context, opts ...RequestOption) error {
+func (r *resultImpl) Next(ctx context.Context, opts ...RequestOption) error {
 	if r.cursor == nil {
 		r.done = true
 		r.messages = nil
@@ -57,6 +67,6 @@ func (r *Result) Next(ctx context.Context, opts ...RequestOption) error {
 	return nil
 }

-func (r *Result) Messages() []*pb.WakuMessageKeyValue {
+func (r *resultImpl) Messages() []*pb.WakuMessageKeyValue {
 	return r.messages
 }
@@ -3,21 +3,21 @@ package utils
 import (
 	"crypto/ecdsa"

-	"github.com/btcsuite/btcd/btcec/v2"
+	"github.com/decred/dcrd/dcrec/secp256k1/v4"
 	"github.com/libp2p/go-libp2p/core/crypto"
 )

 // EcdsaPubKeyToSecp256k1PublicKey converts an `ecdsa.PublicKey` into a libp2p `crypto.Secp256k1PublicKey“
 func EcdsaPubKeyToSecp256k1PublicKey(pubKey *ecdsa.PublicKey) *crypto.Secp256k1PublicKey {
-	xFieldVal := &btcec.FieldVal{}
-	yFieldVal := &btcec.FieldVal{}
+	xFieldVal := &secp256k1.FieldVal{}
+	yFieldVal := &secp256k1.FieldVal{}
 	xFieldVal.SetByteSlice(pubKey.X.Bytes())
 	yFieldVal.SetByteSlice(pubKey.Y.Bytes())
-	return (*crypto.Secp256k1PublicKey)(btcec.NewPublicKey(xFieldVal, yFieldVal))
+	return (*crypto.Secp256k1PublicKey)(secp256k1.NewPublicKey(xFieldVal, yFieldVal))
 }

 // EcdsaPrivKeyToSecp256k1PrivKey converts an `ecdsa.PrivateKey` into a libp2p `crypto.Secp256k1PrivateKey“
 func EcdsaPrivKeyToSecp256k1PrivKey(privKey *ecdsa.PrivateKey) *crypto.Secp256k1PrivateKey {
-	privK, _ := btcec.PrivKeyFromBytes(privKey.D.Bytes())
+	privK := secp256k1.PrivKeyFromBytes(privKey.D.Bytes())
 	return (*crypto.Secp256k1PrivateKey)(privK)
 }
@@ -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
Richard Ramos	1608cf2b0b	fix: populate pinger	2024-11-28 14:38:57 -04:00
Zoro	68a6faaf5c	chore: use decred secp256k1 directly (#1245 )	2024-11-25 15:06:12 -04:00
richΛrd	f98a17bacf	feat: limit api store queries to 24h (#1257 )	2024-11-25 15:05:47 -04:00
richΛrd	96702e278b	fix: setup shards (#1256 )	2024-11-22 09:36:02 -04:00
richΛrd	dd82c24e00	chore: use AddrInfo instead of ID to match nwaku's libwaku ping function (#1251 )	2024-10-28 15:46:39 -04:00
Prem Chaitanya Prathi	c78b09d4ca	fix: filter stats mismatch and add bad peer check for light mode (#1241 )	2024-10-28 11:46:40 +05:30
richΛrd	fdb3c3d0b3	refactor: use protobuffer for API storenode queries (#1248 )	2024-10-24 14:47:57 -04:00
Richard Ramos	6bdf125dd1	refactor: extract ping interface	2024-10-24 14:32:22 -04:00
kaichao	38be0dc169	chore: fix store request id log (#1242 )	2024-10-21 11:23:49 +08:00
richΛrd	37f936d747	refactor: decouple API from go-waku (#1239 )	2024-10-15 15:48:15 -04:00
richΛrd	76275f6fb8	feat: storenode cycle (#1223 )	2024-10-14 14:58:51 -04:00