fix: concurrent peerMap iteration and write (#1266)

cmd/waku/node.go

@@ -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)
 	}

logging/logging.go

@@ -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()))
 }

waku/v2/api/common/pinger.go

@@ -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
+	}
+}

waku/v2/api/common/result.go

@@ -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
+}

waku/v2/api/common/storenode_requestor.go

@@ -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)
+}

waku/v2/api/filter/filter.go

@@ -49,41 +49,34 @@ type Sub struct {
 	errcnt                int
 }
 
-type FilterParameters struct {
+type subscribeParameters struct {
 	batchInterval          time.Duration
 	multiplexChannelBuffer int
-	checker                *onlinechecker.DefaultOnlineChecker
 }
 
-type FilterOptions func(*FilterParameters)
+type SubscribeOptions func(*subscribeParameters)
 
-func WithBatchInterval(t time.Duration) FilterOptions {
-	return func(params *FilterParameters) {
+func WithBatchInterval(t time.Duration) SubscribeOptions {
+	return func(params *subscribeParameters) {
 		params.batchInterval = t
 	}
 }
 
-func WithMultiplexChannelBuffer(value int) FilterOptions {
-	return func(params *FilterParameters) {
+func WithMultiplexChannelBuffer(value int) SubscribeOptions {
+	return func(params *subscribeParameters) {
 		params.multiplexChannelBuffer = value
 	}
 }
 
-func WithOnlineChecker(checker *onlinechecker.DefaultOnlineChecker) FilterOptions {
-	return func(params *FilterParameters) {
-		params.checker = checker
-	}
-}
-
-func defaultOptions() []FilterOptions {
-	return []FilterOptions{
+func defaultOptions() []SubscribeOptions {
+	return []SubscribeOptions{
 		WithBatchInterval(5 * time.Second),
 		WithMultiplexChannelBuffer(100),
 	}
 }
 
 // Subscribe
-func Subscribe(ctx context.Context, wf *filter.WakuFilterLightNode, contentFilter protocol.ContentFilter, config FilterConfig, log *zap.Logger, params *FilterParameters) (*Sub, error) {
+func Subscribe(ctx context.Context, wf *filter.WakuFilterLightNode, contentFilter protocol.ContentFilter, config FilterConfig, log *zap.Logger, params *subscribeParameters) (*Sub, error) {
 	sub := new(Sub)
 	sub.id = uuid.NewString()
 	sub.wf = wf

waku/v2/api/filter/filter_manager.go

@@ -32,7 +32,7 @@ type appFilterMap map[string]filterConfig
 type FilterManager struct {
 	sync.Mutex
 	ctx                    context.Context
-	params                 *FilterParameters
+	params                 *subscribeParameters
 	minPeersPerFilter      int
 	onlineChecker          *onlinechecker.DefaultOnlineChecker
 	filterSubscriptions    map[string]SubDetails // map of aggregated filters to apiSub details
@@ -61,7 +61,7 @@ type EnevelopeProcessor interface {
 	OnNewEnvelope(env *protocol.Envelope) error
 }
 
-func NewFilterManager(ctx context.Context, logger *zap.Logger, minPeersPerFilter int, envProcessor EnevelopeProcessor, node *filter.WakuFilterLightNode, opts ...FilterOptions) *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
@@ -70,21 +70,18 @@ func NewFilterManager(ctx context.Context, logger *zap.Logger, minPeersPerFilter
 	mgr.envProcessor = envProcessor
 	mgr.filterSubscriptions = make(map[string]SubDetails)
 	mgr.node = node
+	mgr.onlineChecker = onlinechecker.NewDefaultOnlineChecker(false).(*onlinechecker.DefaultOnlineChecker)
+	mgr.node.SetOnlineChecker(mgr.onlineChecker)
 	mgr.incompleteFilterBatch = make(map[string]filterConfig)
 	mgr.filterConfigs = make(appFilterMap)
 	mgr.waitingToSubQueue = make(chan filterConfig, 100)
 
 	//parsing the subscribe params only to read the batchInterval passed.
-	mgr.params = new(FilterParameters)
+	mgr.params = new(subscribeParameters)
 	opts = append(defaultOptions(), opts...)
 	for _, opt := range opts {
 		opt(mgr.params)
 	}
-	if mgr.params.checker == nil {
-		mgr.params.checker = onlinechecker.NewDefaultOnlineChecker(false).(*onlinechecker.DefaultOnlineChecker)
-	}
-	mgr.node.SetOnlineChecker(mgr.params.checker)
-
 	mgr.filterSubBatchDuration = mgr.params.batchInterval
 	go mgr.startFilterSubLoop()
 	return mgr
@@ -233,6 +230,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)
 		}

waku/v2/api/filter/filter_test.go

@@ -54,7 +54,7 @@ func (s *FilterApiTestSuite) TestSubscribe() {
 	s.Require().Equal(contentFilter.PubsubTopic, s.TestTopic)
 	ctx, cancel := context.WithCancel(context.Background())
 	s.Log.Info("About to perform API Subscribe()")
-	params := FilterParameters{300 * time.Second, 1024}
+	params := subscribeParameters{300 * time.Second, 1024}
 	apiSub, err := Subscribe(ctx, s.LightNode, contentFilter, apiConfig, s.Log, &params)
 	s.Require().NoError(err)
 	s.Require().Equal(apiSub.ContentFilter, contentFilter)

waku/v2/api/history/cycle.go

@@ -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")
+}

waku/v2/api/history/emitters.go

@@ -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
+}

waku/v2/api/history/emitters_test.go

@@ -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()
+}

waku/v2/api/history/history.go

@@ -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
+}

waku/v2/api/history/history_test.go

@@ -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)
+}

waku/v2/api/history/sort.go

@@ -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)
+}

waku/v2/api/missing/default_requestor.go

@@ -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)
+}

waku/v2/api/missing/missing_messages.go

@@ -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
@@ -35,20 +37,20 @@ type MissingMessageVerifier struct {
 	ctx    context.Context
 	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
 	criteriaInterestMu sync.RWMutex
 
 	C <-chan *protocol.Envelope
 
-	store      *store.WakuStore
 	timesource timesource.Timesource
 	logger     *zap.Logger
 }
 
 // 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 +58,11 @@ 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,
 	}
 }
 
@@ -178,7 +180,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 +214,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 +254,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 +293,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 +327,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
 					}

waku/v2/api/publish/default_publisher.go

@@ -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))
+}

waku/v2/api/publish/default_verifier.go

@@ -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
+}

waku/v2/api/publish/message_check.go

@@ -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
 			}

waku/v2/api/publish/message_check_test.go

@@ -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)

waku/v2/api/publish/message_sender.go

@@ -8,8 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"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"
 )
@@ -37,10 +36,20 @@ 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
 	logger           *zap.Logger
@@ -65,14 +74,13 @@ 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, logger *zap.Logger) (*MessageSender, error) {
 	if publishMethod == UnknownMethod {
 		return nil, errors.New("publish method is required")
 	}
 	return &MessageSender{
 		publishMethod: publishMethod,
-		lightPush:     lightPush,
-		relay:         relay,
+		publisher:     publisher,
 		rateLimiter:   NewPublishRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst),
 		logger:        logger,
 	}, nil
@@ -109,26 +117,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
 		}
@@ -162,9 +167,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)
-	}
-}

waku/v2/api/publish/message_sender_test.go

@@ -40,7 +40,7 @@ func (m *MockMessageSentCheck) Start() {
 }
 
 func TestNewSenderWithUnknownMethod(t *testing.T) {
-	sender, err := NewMessageSender(UnknownMethod, nil, nil, nil)
+	sender, err := NewMessageSender(UnknownMethod, nil, nil)
 	require.NotNil(t, err)
 	require.Nil(t, sender)
 }
@@ -53,7 +53,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, utils.Logger())
 	require.Nil(t, err)
 	require.NotNil(t, sender)
 	require.Nil(t, sender.messageSentCheck)
@@ -78,7 +79,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, utils.Logger())
 
 	check := &MockMessageSentCheck{Messages: make(map[string]map[common.Hash]uint32)}
 	sender.WithMessageSentCheck(check)
@@ -108,7 +110,7 @@ func TestNewSenderWithRelayAndMessageSentCheck(t *testing.T) {
 }
 
 func TestNewSenderWithLightPush(t *testing.T) {
-	sender, err := NewMessageSender(LightPush, nil, nil, nil)
+	sender, err := NewMessageSender(LightPush, nil, nil)
 	require.Nil(t, err)
 	require.NotNil(t, sender)
 	require.Equal(t, LightPush, sender.publishMethod)

waku/v2/node/wakunode2.go

@@ -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
 		}
 	}
 

waku/v2/peermanager/peer_manager.go

@@ -101,7 +101,9 @@ const (
 const maxFailedAttempts = 5
 const prunePeerStoreInterval = 10 * time.Minute
 const peerConnectivityLoopSecs = 15
-const maxConnsToPeerRatio = 5
+const maxConnsToPeerRatio = 3
+const badPeersCleanupInterval = 1 * time.Minute
+const maxDialFailures = 2
 
 // 80% relay peers 20% service peers
 func relayAndServicePeers(maxConnections int) (int, int) {
@@ -256,16 +258,32 @@ func (pm *PeerManager) Start(ctx context.Context) {
 	}
 }
 
+func (pm *PeerManager) removeBadPeers() {
+	if !pm.RelayEnabled {
+		for _, peerID := range pm.host.Peerstore().Peers() {
+			if pm.host.Peerstore().(wps.WakuPeerstore).ConnFailures(peerID) > maxDialFailures {
+				//delete peer from peerStore
+				pm.logger.Debug("removing bad peer due to recurring dial failures", zap.Stringer("peerID", peerID))
+				pm.RemovePeer(peerID)
+			}
+		}
+	}
+}
+
 func (pm *PeerManager) peerStoreLoop(ctx context.Context) {
 	defer utils.LogOnPanic()
 	t := time.NewTicker(prunePeerStoreInterval)
+	t1 := time.NewTicker(badPeersCleanupInterval)
 	defer t.Stop()
+	defer t1.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-t.C:
 			pm.prunePeerStore()
+		case <-t1.C:
+			pm.removeBadPeers()
 		}
 	}
 }
@@ -744,4 +762,9 @@ func (pm *PeerManager) HandleDialError(err error, peerID peer.ID) {
 			pm.logger.Error("failed to emit DialError", zap.Error(emitterErr))
 		}
 	}
+	if !pm.RelayEnabled && pm.host.Peerstore().(wps.WakuPeerstore).ConnFailures(peerID) >= maxDialFailures {
+		//delete peer from peerStore
+		pm.logger.Debug("removing bad peer due to recurring dial failures", zap.Stringer("peerID", peerID))
+		pm.RemovePeer(peerID)
+	}
 }

waku/v2/peermanager/peer_selection.go

@@ -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 {

waku/v2/protocol/store/client.go

@@ -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
 }

waku/v2/protocol/store/client_test.go

@@ -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)
 }

waku/v2/protocol/store/options.go

@@ -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.

waku/v2/protocol/store/result.go

@@ -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
 }

waku/v2/utils/crypto.go

@@ -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)
 }