status-im
/
status-go
mirror of https://github.com/status-im/status-go.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

fix!: code review

This commit is contained in:
Richard Ramos 2024-10-21 16:34:58 -04:00
parent 95b6a17719
commit 441fee76d2
No known key found for this signature in database
GPG Key ID: 1CE87DB518195760
24 changed files with 323 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
eth-node/bridge/geth/waku.go
View File

@ -320,10 +320,6 @@ func (w *GethWakuWrapper) GetActiveStorenode() peer.ID {
panic("not available in WakuV1") panic("not available in WakuV1")
} }
func (w *GethWakuWrapper) OnStorenodeAvailableOneShot() <-chan struct{} {
panic("not available in WakuV1")
}
func (w *GethWakuWrapper) OnStorenodeChanged() <-chan peer.ID { func (w *GethWakuWrapper) OnStorenodeChanged() <-chan peer.ID {
panic("not available in WakuV1") panic("not available in WakuV1")
} }
@ -336,7 +332,7 @@ func (w *GethWakuWrapper) OnStorenodeAvailable() <-chan peer.ID {
panic("not available in WakuV1") panic("not available in WakuV1")
} }
func (w *GethWakuWrapper) WaitForAvailableStoreNode(timeout time.Duration) bool { func (w *GethWakuWrapper) WaitForAvailableStoreNode(ctx context.Context) bool {
return false return false
} }

12
eth-node/bridge/geth/wakuv2.go
View File

@ -314,10 +314,6 @@ func (w *gethWakuV2Wrapper) GetActiveStorenode() peer.ID {
return w.waku.StorenodeCycle.GetActiveStorenode() return w.waku.StorenodeCycle.GetActiveStorenode()
} }
func (w *gethWakuV2Wrapper) OnStorenodeAvailableOneShot() <-chan struct{} {
return w.waku.StorenodeCycle.StorenodeAvailableOneshotEmitter.Subscribe()
}
func (w *gethWakuV2Wrapper) OnStorenodeChanged() <-chan peer.ID { func (w *gethWakuV2Wrapper) OnStorenodeChanged() <-chan peer.ID {
return w.waku.StorenodeCycle.StorenodeChangedEmitter.Subscribe() return w.waku.StorenodeCycle.StorenodeChangedEmitter.Subscribe()
} }
@ -330,8 +326,8 @@ func (w *gethWakuV2Wrapper) OnStorenodeAvailable() <-chan peer.ID {
return w.waku.StorenodeCycle.StorenodeAvailableEmitter.Subscribe() return w.waku.StorenodeCycle.StorenodeAvailableEmitter.Subscribe()
} }
func (w *gethWakuV2Wrapper) WaitForAvailableStoreNode(timeout time.Duration) bool { func (w *gethWakuV2Wrapper) WaitForAvailableStoreNode(ctx context.Context) bool {
return w.waku.StorenodeCycle.WaitForAvailableStoreNode(context.TODO(), timeout) return w.waku.StorenodeCycle.WaitForAvailableStoreNode(ctx)
} }
func (w *gethWakuV2Wrapper) SetStorenodeConfigProvider(c history.StorenodeConfigProvider) { func (w *gethWakuV2Wrapper) SetStorenodeConfigProvider(c history.StorenodeConfigProvider) {
@ -353,8 +349,8 @@ func (w *gethWakuV2Wrapper) ProcessMailserverBatch(
} }
criteria := store.FilterCriteria{ criteria := store.FilterCriteria{
TimeStart: proto.Int64(int64(batch.From) * int64(time.Second)), TimeStart: proto.Int64(batch.From.UnixNano()),
TimeEnd: proto.Int64(int64(batch.To) * int64(time.Second)), TimeEnd: proto.Int64(batch.From.UnixNano()),
ContentFilter: protocol.NewContentFilter(pubsubTopic, contentTopics...), ContentFilter: protocol.NewContentFilter(pubsubTopic, contentTopics...),
} }

13
eth-node/types/waku.go
View File

@ -202,9 +202,6 @@ type Waku interface {
// GetActiveStorenode returns the peer ID of the currently active storenode. It will be empty if no storenode is active // GetActiveStorenode returns the peer ID of the currently active storenode. It will be empty if no storenode is active
GetActiveStorenode() peer.ID GetActiveStorenode() peer.ID
// OnStorenodeAvailableOneShot returns a channel that will be triggered only once when a storenode becomes available
OnStorenodeAvailableOneShot() <-chan struct{}
// OnStorenodeChanged is triggered when a new storenode is promoted to become the active storenode or when the active storenode is removed // OnStorenodeChanged is triggered when a new storenode is promoted to become the active storenode or when the active storenode is removed
OnStorenodeChanged() <-chan peer.ID OnStorenodeChanged() <-chan peer.ID
@ -214,8 +211,8 @@ type Waku interface {
// OnStorenodeAvailable is triggered when there is a new active storenode selected // OnStorenodeAvailable is triggered when there is a new active storenode selected
OnStorenodeAvailable() <-chan peer.ID OnStorenodeAvailable() <-chan peer.ID
// WaitForAvailableStoreNode will wait for a storenode to be available until `timeout` happens // WaitForAvailableStoreNode will wait for a storenode to be available depending on the context
WaitForAvailableStoreNode(timeout time.Duration) bool WaitForAvailableStoreNode(ctx context.Context) bool
// SetStorenodeConfigProvider will set the configuration provider for the storenode cycle // SetStorenodeConfigProvider will set the configuration provider for the storenode cycle
SetStorenodeConfigProvider(c history.StorenodeConfigProvider) SetStorenodeConfigProvider(c history.StorenodeConfigProvider)
@ -240,8 +237,8 @@ type Waku interface {
} }
type MailserverBatch struct { type MailserverBatch struct {
From uint32 From time.Time
To uint32 To time.Time
Cursor string Cursor string
PubsubTopic string PubsubTopic string
Topics []TopicType Topics []TopicType
@ -249,7 +246,7 @@ type MailserverBatch struct {
} }
func (mb *MailserverBatch) Hash() string { func (mb *MailserverBatch) Hash() string {
data := fmt.Sprintf("%d%d%s%s%v%v", mb.From, mb.To, mb.Cursor, mb.PubsubTopic, mb.Topics, mb.ChatIDs) data := fmt.Sprintf("%d%d%s%s%v%v", mb.From.UnixNano(), mb.To.UnixNano(), mb.Cursor, mb.PubsubTopic, mb.Topics, mb.ChatIDs)
hash := sha256.Sum256([]byte(data)) hash := sha256.Sum256([]byte(data))
return hex.EncodeToString(hash[:4]) return hex.EncodeToString(hash[:4])
} }

2
go.mod
View File

@ -96,7 +96,7 @@ require (
github.com/schollz/peerdiscovery v1.7.0 github.com/schollz/peerdiscovery v1.7.0
github.com/siphiuel/lc-proxy-wrapper v0.0.0-20230516150924-246507cee8c7 github.com/siphiuel/lc-proxy-wrapper v0.0.0-20230516150924-246507cee8c7
github.com/urfave/cli/v2 v2.27.2 github.com/urfave/cli/v2 v2.27.2
github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835 github.com/waku-org/go-waku v0.8.1-0.20241021202955-3c4e40c729a0
github.com/wk8/go-ordered-map/v2 v2.1.7 github.com/wk8/go-ordered-map/v2 v2.1.7
github.com/yeqown/go-qrcode/v2 v2.2.1 github.com/yeqown/go-qrcode/v2 v2.2.1
github.com/yeqown/go-qrcode/writer/standard v1.2.1 github.com/yeqown/go-qrcode/writer/standard v1.2.1

4
go.sum
View File

@ -2150,8 +2150,8 @@ github.com/waku-org/go-libp2p-pubsub v0.12.0-gowaku.0.20240823143342-b0f2429ca27
github.com/waku-org/go-libp2p-pubsub v0.12.0-gowaku.0.20240823143342-b0f2429ca27f/go.mod h1:Oi0zw9aw8/Y5GC99zt+Ef2gYAl+0nZlwdJonDyOz/sE= github.com/waku-org/go-libp2p-pubsub v0.12.0-gowaku.0.20240823143342-b0f2429ca27f/go.mod h1:Oi0zw9aw8/Y5GC99zt+Ef2gYAl+0nZlwdJonDyOz/sE=
github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0 h1:R4YYx2QamhBRl/moIxkDCNW+OP7AHbyWLBygDc/xIMo= github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0 h1:R4YYx2QamhBRl/moIxkDCNW+OP7AHbyWLBygDc/xIMo=
github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0/go.mod h1:EhZP9fee0DYjKH/IOQvoNSy1tSHp2iZadsHGphcAJgY= github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0/go.mod h1:EhZP9fee0DYjKH/IOQvoNSy1tSHp2iZadsHGphcAJgY=
github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835 h1:Vp6BhXiDEilmchHy8OLMZVhugudsnvveNkAKD5nhAGk= github.com/waku-org/go-waku v0.8.1-0.20241021202955-3c4e40c729a0 h1:PNKcOPMn0yoC2NQaJPPB8FvHT/YtaU8hZAoovSl42KM=
github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835/go.mod h1:1BRnyg2mQ2aBNLTBaPq6vEvobzywGykPOhGQFbHGf74= github.com/waku-org/go-waku v0.8.1-0.20241021202955-3c4e40c729a0/go.mod h1:1BRnyg2mQ2aBNLTBaPq6vEvobzywGykPOhGQFbHGf74=
github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59 h1:jisj+OCI6QydLtFq3Pyhu49wl9ytPN7oAHjMfepHDrA= github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59 h1:jisj+OCI6QydLtFq3Pyhu49wl9ytPN7oAHjMfepHDrA=
github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59/go.mod h1:1PdBdPzyTaKt3VnpAHk3zj+r9dXPFOr3IHZP9nFle6E= github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59/go.mod h1:1PdBdPzyTaKt3VnpAHk3zj+r9dXPFOr3IHZP9nFle6E=
github.com/waku-org/go-zerokit-rln-apple v0.0.0-20230916172309-ee0ee61dde2b h1:KgZVhsLkxsj5gb/FfndSCQu6VYwALrCOgYI3poR95yE= github.com/waku-org/go-zerokit-rln-apple v0.0.0-20230916172309-ee0ee61dde2b h1:KgZVhsLkxsj5gb/FfndSCQu6VYwALrCOgYI3poR95yE=

8
protocol/messenger.go
View File

@ -862,7 +862,13 @@ func (m *Messenger) Start() (*MessengerResponse, error) {
if m.archiveManager.IsReady() { if m.archiveManager.IsReady() {
go func() { go func() {
defer gocommon.LogOnPanic() defer gocommon.LogOnPanic()
<-m.transport.OnStorenodeAvailableOneShot()
select {
case <-m.ctx.Done():
return
case <-m.transport.OnStorenodeAvailable():
}
m.InitHistoryArchiveTasks(controlledCommunities) m.InitHistoryArchiveTasks(controlledCommunities)
}() }()
} }

16
protocol/messenger_communities.go
View File

@ -3293,7 +3293,7 @@ func (m *Messenger) FetchCommunity(request *FetchCommunityRequest) (*communities
WithWaitForResponseOption(request.WaitForResponse), WithWaitForResponseOption(request.WaitForResponse),
} }
community, _, err := m.storeNodeRequestsManager.FetchCommunity(communityAddress, options) community, _, err := m.storeNodeRequestsManager.FetchCommunity(m.ctx, communityAddress, options)
return community, err return community, err
} }
@ -3301,7 +3301,7 @@ func (m *Messenger) FetchCommunity(request *FetchCommunityRequest) (*communities
// fetchCommunities installs filter for community and requests its details from store node. // fetchCommunities installs filter for community and requests its details from store node.
// When response received it will be passed through signals handler. // When response received it will be passed through signals handler.
func (m *Messenger) fetchCommunities(communities []communities.CommunityShard) error { func (m *Messenger) fetchCommunities(communities []communities.CommunityShard) error {
return m.storeNodeRequestsManager.FetchCommunities(communities, []StoreNodeRequestOption{}) return m.storeNodeRequestsManager.FetchCommunities(m.ctx, communities, []StoreNodeRequestOption{})
} }
// passStoredCommunityInfoToSignalHandler calls signal handler with community info // passStoredCommunityInfoToSignalHandler calls signal handler with community info
@ -3972,7 +3972,7 @@ func (m *Messenger) InitHistoryArchiveTasks(communities []*communities.Community
} }
// Request possibly missed waku messages for community // Request possibly missed waku messages for community
ms := m.getCommunityMailserver(c.ID().String()) ms := m.getCommunityStorenode(c.ID().String())
_, err = m.syncFiltersFrom(ms, filters, uint32(latestWakuMessageTimestamp)) _, err = m.syncFiltersFrom(ms, filters, uint32(latestWakuMessageTimestamp))
if err != nil { if err != nil {
m.logger.Error("failed to request missing messages", zap.Error(err)) m.logger.Error("failed to request missing messages", zap.Error(err))
@ -5158,9 +5158,9 @@ func (m *Messenger) startRequestMissingCommunityChannelsHRKeysLoop() {
}() }()
} }
// getCommunityMailserver returns the active mailserver if a communityID is present then it'll return the mailserver // getCommunityStorenode returns the active mailserver if a communityID is present then it'll return the mailserver
// for that community if it has a mailserver setup otherwise it'll return the global mailserver // for that community if it has a mailserver setup otherwise it'll return the global mailserver
func (m *Messenger) getCommunityMailserver(communityID ...string) peer.ID { func (m *Messenger) getCommunityStorenode(communityID ...string) peer.ID {
if m.transport.WakuVersion() != 2 { if m.transport.WakuVersion() != 2 {
return "" return ""
} }
@ -5178,7 +5178,11 @@ func (m *Messenger) getCommunityMailserver(communityID ...string) peer.ID {
return m.transport.GetActiveStorenode() return m.transport.GetActiveStorenode()
} }
peerID, _ := ms.PeerID() peerID, err := ms.PeerID()
if err != nil {
m.logger.Error("getting storenode for community, using global", zap.String("communityID", communityID[0]), zap.Error(err))
return m.transport.GetActiveStorenode()
}
return peerID return peerID
} }

2
protocol/messenger_contacts.go
View File

@ -1321,7 +1321,7 @@ func (m *Messenger) FetchContact(contactID string, waitForResponse bool) (*Conta
options := []StoreNodeRequestOption{ options := []StoreNodeRequestOption{
WithWaitForResponseOption(waitForResponse), WithWaitForResponseOption(waitForResponse),
} }
contact, _, err := m.storeNodeRequestsManager.FetchContact(contactID, options) contact, _, err := m.storeNodeRequestsManager.FetchContact(m.ctx, contactID, options)
return contact, err return contact, err
} }

79
protocol/messenger_mailserver.go
View File

@ -2,7 +2,6 @@ package protocol
import ( import (
"fmt" "fmt"
"math"
"sort" "sort"
"time" "time"
@ -69,7 +68,7 @@ func (m *Messenger) scheduleSyncChat(chat *Chat) (bool, error) {
go func() { go func() {
defer gocommon.LogOnPanic() defer gocommon.LogOnPanic()
peerID := m.getCommunityMailserver(chat.CommunityID) peerID := m.getCommunityStorenode(chat.CommunityID)
_, err = m.performStorenodeTask(func() (*MessengerResponse, error) { _, err = m.performStorenodeTask(func() (*MessengerResponse, error) {
response, err := m.syncChatWithFilters(peerID, chat.ID) response, err := m.syncChatWithFilters(peerID, chat.ID)
@ -95,6 +94,7 @@ func (m *Messenger) performStorenodeTask(task func() (*MessengerResponse, error)
errCh := make(chan error) errCh := make(chan error)
go func() { go func() {
defer gocommon.LogOnPanic()
err := m.transport.PerformStorenodeTask(func() error { err := m.transport.PerformStorenodeTask(func() error {
r, err := task() r, err := task()
if err != nil { if err != nil {
@ -102,6 +102,8 @@ func (m *Messenger) performStorenodeTask(task func() (*MessengerResponse, error)
} }
select { select {
case <-m.ctx.Done():
return m.ctx.Err()
case responseCh <- r: case responseCh <- r:
default: default:
// //
@ -143,7 +145,7 @@ func (m *Messenger) scheduleSyncFilters(filters []*transport.Filter) (bool, erro
// split filters by community store node so we can request the filters to the correct mailserver // split filters by community store node so we can request the filters to the correct mailserver
filtersByMs := m.SplitFiltersByStoreNode(filters) filtersByMs := m.SplitFiltersByStoreNode(filters)
for communityID, filtersForMs := range filtersByMs { for communityID, filtersForMs := range filtersByMs {
peerID := m.getCommunityMailserver(communityID) peerID := m.getCommunityStorenode(communityID)
_, err := m.performStorenodeTask(func() (*MessengerResponse, error) { _, err := m.performStorenodeTask(func() (*MessengerResponse, error) {
response, err := m.syncFilters(peerID, filtersForMs) response, err := m.syncFilters(peerID, filtersForMs)
@ -166,15 +168,14 @@ func (m *Messenger) scheduleSyncFilters(filters []*transport.Filter) (bool, erro
return true, nil return true, nil
} }
func (m *Messenger) calculateMailserverTo() uint32 { func (m *Messenger) calculateMailserverTo() time.Time {
seconds := float64(m.GetCurrentTimeInMillis()) / 1000 return time.Unix(0, int64(time.Duration(m.GetCurrentTimeInMillis())*time.Millisecond))
return uint32(math.Ceil(seconds))
} }
func (m *Messenger) calculateMailserverTimeBounds(duration time.Duration) (uint32, uint32) { func (m *Messenger) calculateMailserverTimeBounds(duration time.Duration) (time.Time, time.Time) {
now := float64(m.GetCurrentTimeInMillis()) / 1000 now := time.Unix(0, int64(time.Duration(m.GetCurrentTimeInMillis())*time.Millisecond))
to := uint32(math.Ceil(now)) to := now
from := uint32(math.Floor(now)) - uint32(duration.Seconds()) from := now.Add(-duration)
return from, to return from, to
} }
@ -251,7 +252,7 @@ func (m *Messenger) syncBackup() error {
from, to := m.calculateMailserverTimeBounds(oneMonthDuration) from, to := m.calculateMailserverTimeBounds(oneMonthDuration)
batch := types.MailserverBatch{From: from, To: to, Topics: []types.TopicType{filter.ContentTopic}} batch := types.MailserverBatch{From: from, To: to, Topics: []types.TopicType{filter.ContentTopic}}
ms := m.getCommunityMailserver(filter.ChatID) ms := m.getCommunityStorenode(filter.ChatID)
err = m.processMailserverBatch(ms, batch) err = m.processMailserverBatch(ms, batch)
if err != nil { if err != nil {
return err return err
@ -347,7 +348,7 @@ func (m *Messenger) RequestAllHistoricMessages(forceFetchingBackup, withRetries
filtersByMs := m.SplitFiltersByStoreNode(filters) filtersByMs := m.SplitFiltersByStoreNode(filters)
allResponses := &MessengerResponse{} allResponses := &MessengerResponse{}
for communityID, filtersForMs := range filtersByMs { for communityID, filtersForMs := range filtersByMs {
peerID := m.getCommunityMailserver(communityID) peerID := m.getCommunityStorenode(communityID)
if withRetries { if withRetries {
response, err := m.performStorenodeTask(func() (*MessengerResponse, error) { response, err := m.performStorenodeTask(func() (*MessengerResponse, error) {
return m.syncFilters(peerID, filtersForMs) return m.syncFilters(peerID, filtersForMs)
@ -402,7 +403,7 @@ func (m *Messenger) checkForMissingMessagesLoop() {
filters := m.transport.Filters() filters := m.transport.Filters()
filtersByMs := m.SplitFiltersByStoreNode(filters) filtersByMs := m.SplitFiltersByStoreNode(filters)
for communityID, filtersForMs := range filtersByMs { for communityID, filtersForMs := range filtersByMs {
peerID := m.getCommunityMailserver(communityID) peerID := m.getCommunityStorenode(communityID)
if peerID == "" { if peerID == "" {
continue continue
} }
@ -533,7 +534,7 @@ func (m *Messenger) syncFiltersFrom(peerID peer.ID, filters []*transport.Filter,
return nil, err return nil, err
} }
} }
batch = types.MailserverBatch{From: from, To: to} batch = types.MailserverBatch{From: time.Unix(int64(from), 0), To: to}
} }
batch.ChatIDs = append(batch.ChatIDs, chatID) batch.ChatIDs = append(batch.ChatIDs, chatID)
@ -542,7 +543,7 @@ func (m *Messenger) syncFiltersFrom(peerID peer.ID, filters []*transport.Filter,
batches[pubsubTopic][batchID] = batch batches[pubsubTopic][batchID] = batch
// Set last request to the new `to` // Set last request to the new `to`
topicData.LastRequest = int(to) topicData.LastRequest = int(to.Unix())
syncedTopics = append(syncedTopics, topicData) syncedTopics = append(syncedTopics, topicData)
} }
} }
@ -574,8 +575,8 @@ func (m *Messenger) syncFiltersFrom(peerID peer.ID, filters []*transport.Filter,
ChatIDs: batch.ChatIDs, ChatIDs: batch.ChatIDs,
} }
from := batch.To - uint32(oneDayDuration.Seconds()) from := batch.To.Add(-oneDayDuration)
if from > batch.From { if from.After(batch.From) {
dayBatch.From = from dayBatch.From = from
batches24h = append(batches24h, dayBatch) batches24h = append(batches24h, dayBatch)
@ -621,15 +622,15 @@ func (m *Messenger) syncFiltersFrom(peerID peer.ID, filters []*transport.Filter,
if !ok || !chat.Active || chat.Timeline() || chat.ProfileUpdates() { if !ok || !chat.Active || chat.Timeline() || chat.ProfileUpdates() {
continue continue
} }
gap, err := m.calculateGapForChat(chat, batch.From) gap, err := m.calculateGapForChat(chat, uint32(batch.From.Unix()))
if err != nil { if err != nil {
return nil, err return nil, err
} }
if chat.SyncedFrom == 0 || chat.SyncedFrom > batch.From { if chat.SyncedFrom == 0 || chat.SyncedFrom > uint32(batch.From.Unix()) {
chat.SyncedFrom = batch.From chat.SyncedFrom = uint32(batch.From.Unix())
} }
chat.SyncedTo = to chat.SyncedTo = uint32(to.Unix())
err = m.persistence.SetSyncTimestamps(chat.SyncedFrom, chat.SyncedTo, chat.ID) err = m.persistence.SetSyncTimestamps(chat.SyncedFrom, chat.SyncedTo, chat.ID)
if err != nil { if err != nil {
@ -739,7 +740,7 @@ func (m *Messenger) SyncChatFromSyncedFrom(chatID string) (uint32, error) {
return 0, ErrChatNotFound return 0, ErrChatNotFound
} }
peerID := m.getCommunityMailserver(chat.CommunityID) peerID := m.getCommunityStorenode(chat.CommunityID)
var from uint32 var from uint32
_, err := m.performStorenodeTask(func() (*MessengerResponse, error) { _, err := m.performStorenodeTask(func() (*MessengerResponse, error) {
canSync, err := m.canSyncWithStoreNodes() canSync, err := m.canSyncWithStoreNodes()
@ -762,8 +763,8 @@ func (m *Messenger) SyncChatFromSyncedFrom(chatID string) (uint32, error) {
batch := types.MailserverBatch{ batch := types.MailserverBatch{
ChatIDs: []string{chatID}, ChatIDs: []string{chatID},
To: chat.SyncedFrom, To: time.Unix(int64(chat.SyncedFrom), 0),
From: chat.SyncedFrom - defaultSyncPeriod, From: time.Unix(int64(chat.SyncedFrom-defaultSyncPeriod), 0),
PubsubTopic: pubsubTopic, PubsubTopic: pubsubTopic,
Topics: topics, Topics: topics,
} }
@ -779,14 +780,14 @@ func (m *Messenger) SyncChatFromSyncedFrom(chatID string) (uint32, error) {
if m.config.messengerSignalsHandler != nil { if m.config.messengerSignalsHandler != nil {
m.config.messengerSignalsHandler.HistoryRequestCompleted() m.config.messengerSignalsHandler.HistoryRequestCompleted()
} }
if chat.SyncedFrom == 0 || chat.SyncedFrom > batch.From { if chat.SyncedFrom == 0 || chat.SyncedFrom > uint32(batch.From.Unix()) {
chat.SyncedFrom = batch.From chat.SyncedFrom = uint32(batch.From.Unix())
} }
m.logger.Debug("setting sync timestamps", zap.Int64("from", int64(batch.From)), zap.Int64("to", int64(chat.SyncedTo)), zap.String("chatID", chatID)) m.logger.Debug("setting sync timestamps", zap.Int64("from", int64(batch.From.Unix())), zap.Int64("to", int64(chat.SyncedTo)), zap.String("chatID", chatID))
err = m.persistence.SetSyncTimestamps(batch.From, chat.SyncedTo, chat.ID) err = m.persistence.SetSyncTimestamps(uint32(batch.From.Unix()), chat.SyncedTo, chat.ID)
from = batch.From from = uint32(batch.From.Unix())
return nil, err return nil, err
}, history.WithPeerID(peerID)) }, history.WithPeerID(peerID))
if err != nil { if err != nil {
@ -830,8 +831,8 @@ func (m *Messenger) FillGaps(chatID string, messageIDs []string) error {
batch := types.MailserverBatch{ batch := types.MailserverBatch{
ChatIDs: []string{chatID}, ChatIDs: []string{chatID},
To: highestTo, To: time.Unix(int64(highestTo), 0),
From: lowestFrom, From: time.Unix(int64(lowestFrom), 0),
PubsubTopic: pubsubTopic, PubsubTopic: pubsubTopic,
Topics: topics, Topics: topics,
} }
@ -840,7 +841,7 @@ func (m *Messenger) FillGaps(chatID string, messageIDs []string) error {
m.config.messengerSignalsHandler.HistoryRequestStarted(1) m.config.messengerSignalsHandler.HistoryRequestStarted(1)
} }
peerID := m.getCommunityMailserver(chat.CommunityID) peerID := m.getCommunityStorenode(chat.CommunityID)
err = m.processMailserverBatch(peerID, batch) err = m.processMailserverBatch(peerID, batch)
if err != nil { if err != nil {
return err return err
@ -907,7 +908,7 @@ func (m *Messenger) fetchMessages(chatID string, duration time.Duration) (uint32
return 0, ErrChatNotFound return 0, ErrChatNotFound
} }
peerID := m.getCommunityMailserver(chat.CommunityID) peerID := m.getCommunityStorenode(chat.CommunityID)
_, err := m.performStorenodeTask(func() (*MessengerResponse, error) { _, err := m.performStorenodeTask(func() (*MessengerResponse, error) {
canSync, err := m.canSyncWithStoreNodes() canSync, err := m.canSyncWithStoreNodes()
if err != nil { if err != nil {
@ -917,7 +918,7 @@ func (m *Messenger) fetchMessages(chatID string, duration time.Duration) (uint32
return nil, nil return nil, nil
} }
m.logger.Debug("fetching messages", zap.String("chatID", chatID), zap.Stringer("storenodeID", peerID)) m.logger.Debug("fetching messages", zap.String("chatID", chatID), zap.Stringer("peerID", peerID))
pubsubTopic, topics, err := m.topicsForChat(chatID) pubsubTopic, topics, err := m.topicsForChat(chatID)
if err != nil { if err != nil {
return nil, nil return nil, nil
@ -942,13 +943,13 @@ func (m *Messenger) fetchMessages(chatID string, duration time.Duration) (uint32
if m.config.messengerSignalsHandler != nil { if m.config.messengerSignalsHandler != nil {
m.config.messengerSignalsHandler.HistoryRequestCompleted() m.config.messengerSignalsHandler.HistoryRequestCompleted()
} }
if chat.SyncedFrom == 0 || chat.SyncedFrom > batch.From { if chat.SyncedFrom == 0 || chat.SyncedFrom > uint32(batch.From.Second()) {
chat.SyncedFrom = batch.From chat.SyncedFrom = uint32(batch.From.Second())
} }
m.logger.Debug("setting sync timestamps", zap.Int64("from", int64(batch.From)), zap.Int64("to", int64(chat.SyncedTo)), zap.String("chatID", chatID)) m.logger.Debug("setting sync timestamps", zap.Int64("from", batch.From.Unix()), zap.Int64("to", int64(chat.SyncedTo)), zap.String("chatID", chatID))
err = m.persistence.SetSyncTimestamps(batch.From, chat.SyncedTo, chat.ID) err = m.persistence.SetSyncTimestamps(uint32(batch.From.Unix()), chat.SyncedTo, chat.ID)
from = batch.From from = batch.From
return nil, err return nil, err
}, history.WithPeerID(peerID)) }, history.WithPeerID(peerID))
@ -956,5 +957,5 @@ func (m *Messenger) fetchMessages(chatID string, duration time.Duration) (uint32
return 0, err return 0, err
} }
return from, nil return uint32(from.Unix()), nil
} }

29
protocol/messenger_store_node_request_manager.go
View File

@ -1,6 +1,7 @@
package protocol package protocol
import ( import (
"context"
"database/sql" "database/sql"
"fmt" "fmt"
"strings" "strings"
@ -75,7 +76,7 @@ func NewStoreNodeRequestManager(m *Messenger) *StoreNodeRequestManager {
// the function will also wait for the store node response and return the fetched community. // the function will also wait for the store node response and return the fetched community.
// Automatically waits for an available store node. // Automatically waits for an available store node.
// When a `nil` community and `nil` error is returned, that means the community wasn't found at the store node. // When a `nil` community and `nil` error is returned, that means the community wasn't found at the store node.
func (m *StoreNodeRequestManager) FetchCommunity(community communities.CommunityShard, opts []StoreNodeRequestOption) (*communities.Community, StoreNodeRequestStats, error) { func (m *StoreNodeRequestManager) FetchCommunity(ctx context.Context, community communities.CommunityShard, opts []StoreNodeRequestOption) (*communities.Community, StoreNodeRequestStats, error) {
cfg := buildStoreNodeRequestConfig(opts) cfg := buildStoreNodeRequestConfig(opts)
m.logger.Info("requesting community from store node", m.logger.Info("requesting community from store node",
@ -83,7 +84,7 @@ func (m *StoreNodeRequestManager) FetchCommunity(community communities.Community
zap.Any("config", cfg)) zap.Any("config", cfg))
requestCommunity := func(communityID string, shard *shard.Shard) (*communities.Community, StoreNodeRequestStats, error) { requestCommunity := func(communityID string, shard *shard.Shard) (*communities.Community, StoreNodeRequestStats, error) {
channel, err := m.subscribeToRequest(storeNodeCommunityRequest, communityID, shard, cfg) channel, err := m.subscribeToRequest(ctx, storeNodeCommunityRequest, communityID, shard, cfg)
if err != nil { if err != nil {
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err) return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err)
} }
@ -100,7 +101,7 @@ func (m *StoreNodeRequestManager) FetchCommunity(community communities.Community
communityShard := community.Shard communityShard := community.Shard
if communityShard == nil { if communityShard == nil {
id := transport.CommunityShardInfoTopic(community.CommunityID) id := transport.CommunityShardInfoTopic(community.CommunityID)
fetchedShard, err := m.subscribeToRequest(storeNodeShardRequest, id, shard.DefaultNonProtectedShard(), cfg) fetchedShard, err := m.subscribeToRequest(ctx, storeNodeShardRequest, id, shard.DefaultNonProtectedShard(), cfg)
if err != nil { if err != nil {
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a shard info request: %w", err) return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a shard info request: %w", err)
} }
@ -134,7 +135,7 @@ func (m *StoreNodeRequestManager) FetchCommunity(community communities.Community
// those content topics is spammed with to many envelopes, then on each iteration we will have to fetch all // those content topics is spammed with to many envelopes, then on each iteration we will have to fetch all
// of this spam first to get the envelopes in other content topics. To avoid this we keep independent requests // of this spam first to get the envelopes in other content topics. To avoid this we keep independent requests
// for each content topic. // for each content topic.
func (m *StoreNodeRequestManager) FetchCommunities(communities []communities.CommunityShard, opts []StoreNodeRequestOption) error { func (m *StoreNodeRequestManager) FetchCommunities(ctx context.Context, communities []communities.CommunityShard, opts []StoreNodeRequestOption) error {
m.logger.Info("requesting communities from store node", zap.Any("communities", communities)) m.logger.Info("requesting communities from store node", zap.Any("communities", communities))
// when fetching multiple communities we don't wait for the response // when fetching multiple communities we don't wait for the response
@ -143,7 +144,7 @@ func (m *StoreNodeRequestManager) FetchCommunities(communities []communities.Com
var outErr error var outErr error
for _, community := range communities { for _, community := range communities {
_, _, err := m.FetchCommunity(community, opts) _, _, err := m.FetchCommunity(ctx, community, opts)
if err != nil { if err != nil {
outErr = fmt.Errorf("%sfailed to create a request for community %s: %w", outErr, community.CommunityID, err) outErr = fmt.Errorf("%sfailed to create a request for community %s: %w", outErr, community.CommunityID, err)
} }
@ -154,7 +155,7 @@ func (m *StoreNodeRequestManager) FetchCommunities(communities []communities.Com
// FetchContact - similar to FetchCommunity // FetchContact - similar to FetchCommunity
// If a `nil` contact and a `nil` error are returned, it means that the contact wasn't found at the store node. // If a `nil` contact and a `nil` error are returned, it means that the contact wasn't found at the store node.
func (m *StoreNodeRequestManager) FetchContact(contactID string, opts []StoreNodeRequestOption) (*Contact, StoreNodeRequestStats, error) { func (m *StoreNodeRequestManager) FetchContact(ctx context.Context, contactID string, opts []StoreNodeRequestOption) (*Contact, StoreNodeRequestStats, error) {
cfg := buildStoreNodeRequestConfig(opts) cfg := buildStoreNodeRequestConfig(opts)
@ -162,7 +163,7 @@ func (m *StoreNodeRequestManager) FetchContact(contactID string, opts []StoreNod
zap.Any("contactID", contactID), zap.Any("contactID", contactID),
zap.Any("config", cfg)) zap.Any("config", cfg))
channel, err := m.subscribeToRequest(storeNodeContactRequest, contactID, nil, cfg) channel, err := m.subscribeToRequest(ctx, storeNodeContactRequest, contactID, nil, cfg)
if err != nil { if err != nil {
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err) return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err)
} }
@ -178,7 +179,7 @@ func (m *StoreNodeRequestManager) FetchContact(contactID string, opts []StoreNod
// subscribeToRequest checks if a request for given community/contact is already in progress, creates and installs // subscribeToRequest checks if a request for given community/contact is already in progress, creates and installs
// a new one if not found, and returns a subscription to the result of the found/started request. // a new one if not found, and returns a subscription to the result of the found/started request.
// The subscription can then be used to get the result of the request, this could be either a community/contact or an error. // The subscription can then be used to get the result of the request, this could be either a community/contact or an error.
func (m *StoreNodeRequestManager) subscribeToRequest(requestType storeNodeRequestType, dataID string, shard *shard.Shard, cfg StoreNodeRequestConfig) (storeNodeResponseSubscription, error) { func (m *StoreNodeRequestManager) subscribeToRequest(ctx context.Context, requestType storeNodeRequestType, dataID string, shard *shard.Shard, cfg StoreNodeRequestConfig) (storeNodeResponseSubscription, error) {
// It's important to unlock only after getting the subscription channel. // It's important to unlock only after getting the subscription channel.
// We also lock `activeRequestsLock` during finalizing the requests. This ensures that the subscription // We also lock `activeRequestsLock` during finalizing the requests. This ensures that the subscription
// created in this function will get the result even if the requests proceeds faster than this function ends. // created in this function will get the result even if the requests proceeds faster than this function ends.
@ -206,7 +207,7 @@ func (m *StoreNodeRequestManager) subscribeToRequest(requestType storeNodeReques
return nil, fmt.Errorf("failed to create community filter: %w", err) return nil, fmt.Errorf("failed to create community filter: %w", err)
} }
request = m.newStoreNodeRequest() request = m.newStoreNodeRequest(ctx)
request.config = cfg request.config = cfg
request.pubsubTopic = filter.PubsubTopic request.pubsubTopic = filter.PubsubTopic
request.requestID = requestID request.requestID = requestID
@ -223,9 +224,10 @@ func (m *StoreNodeRequestManager) subscribeToRequest(requestType storeNodeReques
} }
// newStoreNodeRequest creates a new storeNodeRequest struct // newStoreNodeRequest creates a new storeNodeRequest struct
func (m *StoreNodeRequestManager) newStoreNodeRequest() *storeNodeRequest { func (m *StoreNodeRequestManager) newStoreNodeRequest(ctx context.Context) *storeNodeRequest {
return &storeNodeRequest{ return &storeNodeRequest{
manager: m, manager: m,
ctx: ctx,
subscriptions: make([]storeNodeResponseSubscription, 0), subscriptions: make([]storeNodeResponseSubscription, 0),
} }
} }
@ -306,6 +308,7 @@ const (
// For a valid storeNodeRequest to be performed, the user must set all the struct fields and call start method. // For a valid storeNodeRequest to be performed, the user must set all the struct fields and call start method.
type storeNodeRequest struct { type storeNodeRequest struct {
requestID storeNodeRequestID requestID storeNodeRequestID
ctx context.Context
// request parameters // request parameters
pubsubTopic string pubsubTopic string
@ -524,13 +527,15 @@ func (r *storeNodeRequest) routine() {
communityID := r.requestID.getCommunityID() communityID := r.requestID.getCommunityID()
if r.requestID.RequestType != storeNodeCommunityRequest || !r.manager.messenger.communityStorenodes.HasStorenodeSetup(communityID) { if r.requestID.RequestType != storeNodeCommunityRequest || !r.manager.messenger.communityStorenodes.HasStorenodeSetup(communityID) {
if !r.manager.messenger.transport.WaitForAvailableStoreNode(storeNodeAvailableTimeout) { ctx, cancel := context.WithTimeout(r.ctx, storeNodeAvailableTimeout)
defer cancel()
if !r.manager.messenger.transport.WaitForAvailableStoreNode(ctx) {
r.result.err = fmt.Errorf("store node is not available") r.result.err = fmt.Errorf("store node is not available")
return return
} }
} }
storeNode := r.manager.messenger.getCommunityMailserver(communityID) storeNode := r.manager.messenger.getCommunityStorenode(communityID)
// Check if community already exists locally and get Clock. // Check if community already exists locally and get Clock.
if r.requestID.RequestType == storeNodeCommunityRequest { if r.requestID.RequestType == storeNodeCommunityRequest {

4
protocol/messenger_storenode_request_test.go
View File

@ -310,7 +310,9 @@ func (s *MessengerStoreNodeRequestSuite) fetchProfile(m *Messenger, contactID st
} }
func (s *MessengerStoreNodeRequestSuite) WaitForAvailableStoreNode(messenger *Messenger) { func (s *MessengerStoreNodeRequestSuite) WaitForAvailableStoreNode(messenger *Messenger) {
WaitForAvailableStoreNode(&s.Suite, messenger, storeNodeConnectTimeout) ctx, cancel := context.WithTimeout(context.TODO(), storeNodeConnectTimeout)
defer cancel()
WaitForAvailableStoreNode(&s.Suite, messenger, ctx)
} }
func (s *MessengerStoreNodeRequestSuite) setupEnvelopesWatcher(wakuNode *waku2.Waku, topic *wakuV2common.TopicType, cb func(envelope *wakuV2common.ReceivedMessage)) { func (s *MessengerStoreNodeRequestSuite) setupEnvelopesWatcher(wakuNode *waku2.Waku, topic *wakuV2common.TopicType, cb func(envelope *wakuV2common.ReceivedMessage)) {

4
protocol/messenger_testing_utils.go
View File

@ -364,8 +364,8 @@ func SetIdentityImagesAndWaitForChange(s *suite.Suite, messenger *Messenger, tim
s.Require().True(ok) s.Require().True(ok)
} }
func WaitForAvailableStoreNode(s *suite.Suite, m *Messenger, timeout time.Duration) { func WaitForAvailableStoreNode(s *suite.Suite, m *Messenger, ctx context.Context) {
available := m.transport.WaitForAvailableStoreNode(timeout) available := m.transport.WaitForAvailableStoreNode(ctx)
s.Require().True(available) s.Require().True(available)
} }

6
protocol/storenodes/storenodes.go
View File

@ -62,10 +62,8 @@ func (m *CommunityStorenodes) IsCommunityStoreNode(peerID peer.ID) bool {
for _, data := range m.storenodesByCommunityID { for _, data := range m.storenodesByCommunityID {
for _, snode := range data.storenodes { for _, snode := range data.storenodes {
commStorenodeID, err := utils.GetPeerID(snode.Address) commStorenodeID, err := utils.GetPeerID(snode.Address)
if err == nil { if err == nil && commStorenodeID == peerID {
if commStorenodeID == peerID { return true
return true
}
} }
} }
} }

8
protocol/transport/transport.go
View File

@ -639,10 +639,6 @@ func (t *Transport) DisconnectActiveStorenode(ctx context.Context, backoffReason
t.waku.DisconnectActiveStorenode(ctx, backoffReason, shouldCycle) t.waku.DisconnectActiveStorenode(ctx, backoffReason, shouldCycle)
} }
func (t *Transport) OnStorenodeAvailableOneShot() <-chan struct{} {
return t.waku.OnStorenodeAvailableOneShot()
}
func (t *Transport) OnStorenodeChanged() <-chan peer.ID { func (t *Transport) OnStorenodeChanged() <-chan peer.ID {
return t.waku.OnStorenodeChanged() return t.waku.OnStorenodeChanged()
} }
@ -655,8 +651,8 @@ func (t *Transport) OnStorenodeAvailable() <-chan peer.ID {
return t.waku.OnStorenodeAvailable() return t.waku.OnStorenodeAvailable()
} }
func (t *Transport) WaitForAvailableStoreNode(timeout time.Duration) bool { func (t *Transport) WaitForAvailableStoreNode(ctx context.Context) bool {
return t.waku.WaitForAvailableStoreNode(timeout) return t.waku.WaitForAvailableStoreNode(ctx)
} }
func (t *Transport) IsStorenodeAvailable(peerID peer.ID) bool { func (t *Transport) IsStorenodeAvailable(peerID peer.ID) bool {

17
vendor/github.com/waku-org/go-waku/waku/v2/api/common/result.go generated vendored Normal file
View File

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

24
vendor/github.com/waku-org/go-waku/waku/v2/api/history/cycle.go generated vendored
View File

@ -409,14 +409,10 @@ func (m *StorenodeCycle) SetStorenodeConfigProvider(provider StorenodeConfigProv
m.storenodeConfigProvider = provider m.storenodeConfigProvider = provider
} }
func (m *StorenodeCycle) WaitForAvailableStoreNode(ctx context.Context, timeout time.Duration) bool { func (m *StorenodeCycle) WaitForAvailableStoreNode(ctx context.Context) bool {
// Add 1 second to timeout, because the storenode cycle has 1 second ticker, which doesn't tick on start. // 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. // 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 // NOTE: https://stackoverflow.com/questions/32705582/how-to-get-time-tick-to-tick-immediately
timeout += time.Second
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
wg.Add(1) wg.Add(1)
@ -426,7 +422,18 @@ func (m *StorenodeCycle) WaitForAvailableStoreNode(ctx context.Context, timeout
select { select {
case <-m.StorenodeAvailableOneshotEmitter.Subscribe(): case <-m.StorenodeAvailableOneshotEmitter.Subscribe():
case <-ctx.Done(): 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 return
} }
} }
}() }()
@ -434,6 +441,11 @@ func (m *StorenodeCycle) WaitForAvailableStoreNode(ctx context.Context, timeout
select { select {
case <-waitForWaitGroup(&wg): case <-waitForWaitGroup(&wg):
case <-ctx.Done(): 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) return m.IsStorenodeAvailable(m.activeStorenode)

33
vendor/github.com/waku-org/go-waku/waku/v2/api/missing/default_requestor.go generated vendored Normal file
View File

@ -0,0 +1,33 @@
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"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/store"
)
func NewDefaultStorenodeRequestor(store *store.WakuStore) 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) QueryWithCriteria(ctx context.Context, peerID peer.ID, pageSize uint64, pubsubTopic string, contentTopics []string, from *int64, to *int64) (common.StoreRequestResult, error) {
return d.store.Query(ctx, store.FilterCriteria{
ContentFilter: protocol.NewContentFilter(pubsubTopic, contentTopics...),
TimeStart: from,
TimeEnd: to,
}, store.WithPeer(peerID), store.WithPaging(false, pageSize), store.IncludeData(false))
}

50
vendor/github.com/waku-org/go-waku/waku/v2/api/missing/missing_messages.go generated vendored
View File

@ -11,9 +11,9 @@ import (
"github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/logging" "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"
"github.com/waku-org/go-waku/waku/v2/protocol/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/timesource" "github.com/waku-org/go-waku/waku/v2/timesource"
"github.com/waku-org/go-waku/waku/v2/utils" "github.com/waku-org/go-waku/waku/v2/utils"
"go.uber.org/zap" "go.uber.org/zap"
@ -22,6 +22,7 @@ import (
const maxContentTopicsPerRequest = 10 const maxContentTopicsPerRequest = 10
const maxMsgHashesPerRequest = 50 const maxMsgHashesPerRequest = 50
const messageFetchPageSize = 100
// MessageTracker should keep track of messages it has seen before and // MessageTracker should keep track of messages it has seen before and
// provide a way to determine whether a message exists or not. This // provide a way to determine whether a message exists or not. This
@ -30,25 +31,30 @@ type MessageTracker interface {
MessageExists(pb.MessageHash) (bool, error) MessageExists(pb.MessageHash) (bool, error)
} }
type StorenodeRequestor interface {
GetMessagesByHash(ctx context.Context, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) (common.StoreRequestResult, error)
QueryWithCriteria(ctx context.Context, peerID peer.ID, pageSize uint64, pubsubTopic string, contentTopics []string, from *int64, to *int64) (common.StoreRequestResult, error)
}
// MissingMessageVerifier is used to periodically retrieve missing messages from store nodes that have some specific criteria // MissingMessageVerifier is used to periodically retrieve missing messages from store nodes that have some specific criteria
type MissingMessageVerifier struct { type MissingMessageVerifier struct {
ctx context.Context ctx context.Context
params missingMessageVerifierParams params missingMessageVerifierParams
messageTracker MessageTracker storenodeRequestor 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 criteriaInterestMu sync.RWMutex
C <-chan *protocol.Envelope C <-chan *protocol.Envelope
store *store.WakuStore
timesource timesource.Timesource timesource timesource.Timesource
logger *zap.Logger logger *zap.Logger
} }
// NewMissingMessageVerifier creates an instance of a MissingMessageVerifier // 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 StorenodeRequestor, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier {
options = append(defaultMissingMessagesVerifierOptions, options...) options = append(defaultMissingMessagesVerifierOptions, options...)
params := missingMessageVerifierParams{} params := missingMessageVerifierParams{}
for _, opt := range options { for _, opt := range options {
@ -56,11 +62,11 @@ func NewMissingMessageVerifier(store *store.WakuStore, messageTracker MessageTra
} }
return &MissingMessageVerifier{ return &MissingMessageVerifier{
store: store, storenodeRequestor: storenodeRequester,
timesource: timesource, timesource: timesource,
messageTracker: messageTracker, messageTracker: messageTracker,
logger: logger.Named("missing-msg-verifier"), logger: logger.Named("missing-msg-verifier"),
params: params, params: params,
} }
} }
@ -178,7 +184,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 retry := true
count := 1 count := 1
for retry && count <= m.params.maxAttemptsToRetrieveHistory { for retry && count <= m.params.maxAttemptsToRetrieveHistory {
@ -212,12 +218,16 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
logging.Epoch("to", now), logging.Epoch("to", now),
) )
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) {
return m.store.Query(ctx, store.FilterCriteria{ return m.storenodeRequestor.QueryWithCriteria(
ContentFilter: protocol.NewContentFilter(interest.contentFilter.PubsubTopic, contentTopics[batchFrom:batchTo]...), ctx,
TimeStart: proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()), interest.peerID,
TimeEnd: proto.Int64(now.Add(-m.params.delay).UnixNano()), messageFetchPageSize,
}, store.WithPeer(interest.peerID), store.WithPaging(false, 100), store.IncludeData(false)) interest.contentFilter.PubsubTopic,
contentTopics[batchFrom:batchTo],
proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()),
proto.Int64(now.Add(-m.params.delay).UnixNano()),
)
}, logger, "retrieving history to check for missing messages") }, logger, "retrieving history to check for missing messages")
if err != nil { if err != nil {
if !errors.Is(err, context.Canceled) { if !errors.Is(err, context.Canceled) {
@ -243,7 +253,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
missingHashes = append(missingHashes, hash) 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 { if err = result.Next(ctx); err != nil {
return nil, err return nil, err
} }
@ -282,10 +292,10 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
defer utils.LogOnPanic() defer utils.LogOnPanic()
defer wg.Wait() 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) queryCtx, cancel := context.WithTimeout(ctx, m.params.storeQueryTimeout)
defer cancel() defer cancel()
return m.store.QueryByHash(queryCtx, messageHashes, store.WithPeer(interest.peerID), store.WithPaging(false, maxMsgHashesPerRequest)) return m.storenodeRequestor.GetMessagesByHash(queryCtx, interest.peerID, maxMsgHashesPerRequest, messageHashes)
}, logger, "retrieving missing messages") }, logger, "retrieving missing messages")
if err != nil { if err != nil {
if !errors.Is(err, context.Canceled) { if !errors.Is(err, context.Canceled) {
@ -303,7 +313,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 { if err = result.Next(ctx); err != nil {
return nil, err return nil, err
} }

50
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_publisher.go generated vendored Normal file
View File

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

39
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_verifier.go generated vendored Normal file
View File

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

26
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_check.go generated vendored
View File

@ -8,11 +8,11 @@ import (
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil" "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" 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/api/history"
"github.com/waku-org/go-waku/waku/v2/protocol" "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/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/timesource"
"github.com/waku-org/go-waku/waku/v2/utils" "github.com/waku-org/go-waku/waku/v2/utils"
"go.uber.org/zap" "go.uber.org/zap"
@ -31,6 +31,11 @@ type ISentCheck interface {
DeleteByMessageIDs(messageIDs []common.Hash) DeleteByMessageIDs(messageIDs []common.Hash)
} }
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 // MessageSentCheck tracks the outgoing messages and check against store node
// if the message sent time has passed the `messageSentPeriod`, the message id will be includes for the next query // if the message sent time has passed the `messageSentPeriod`, the message id will be includes for the next query
// if the message keeps missing after `messageExpiredPerid`, the message id will be expired // if the message keeps missing after `messageExpiredPerid`, the message id will be expired
@ -40,7 +45,7 @@ type MessageSentCheck struct {
messageStoredChan chan common.Hash messageStoredChan chan common.Hash
messageExpiredChan chan common.Hash messageExpiredChan chan common.Hash
ctx context.Context ctx context.Context
store *store.WakuStore messageVerifier StorenodeMessageVerifier
storenodeCycle *history.StorenodeCycle storenodeCycle *history.StorenodeCycle
timesource timesource.Timesource timesource timesource.Timesource
logger *zap.Logger logger *zap.Logger
@ -52,14 +57,14 @@ type MessageSentCheck struct {
} }
// NewMessageSentCheck creates a new instance of MessageSentCheck with default parameters // NewMessageSentCheck creates a new instance of MessageSentCheck with default parameters
func NewMessageSentCheck(ctx context.Context, store *store.WakuStore, cycle *history.StorenodeCycle, 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{ return &MessageSentCheck{
messageIDs: make(map[string]map[common.Hash]uint32), messageIDs: make(map[string]map[common.Hash]uint32),
messageIDsMu: sync.RWMutex{}, messageIDsMu: sync.RWMutex{},
messageStoredChan: msgStoredChan, messageStoredChan: msgStoredChan,
messageExpiredChan: msgExpiredChan, messageExpiredChan: msgExpiredChan,
ctx: ctx, ctx: ctx,
store: store, messageVerifier: messageVerifier,
storenodeCycle: cycle, storenodeCycle: cycle,
timesource: timesource, timesource: timesource,
logger: logger, logger: logger,
@ -212,12 +217,7 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c
return []common.Hash{} return []common.Hash{}
} }
var opts []store.RequestOption
requestID := protocol.GenerateRequestID() 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)) messageHashes := make([]pb.MessageHash, len(hashes))
for i, hash := range hashes { for i, hash := range hashes {
@ -228,20 +228,20 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c
queryCtx, cancel := context.WithTimeout(ctx, m.storeQueryTimeout) queryCtx, cancel := context.WithTimeout(ctx, m.storeQueryTimeout)
defer cancel() defer cancel()
result, err := m.store.QueryByHash(queryCtx, messageHashes, opts...) result, err := m.messageVerifier.MessageHashesExist(queryCtx, requestID, selectedPeer, m.maxHashQueryLength, messageHashes)
if err != nil { if err != nil {
m.logger.Error("store.queryByHash failed", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer), zap.Error(err)) m.logger.Error("store.queryByHash failed", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer), zap.Error(err))
return []common.Hash{} 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 ackHashes []common.Hash
var missedHashes []common.Hash var missedHashes []common.Hash
for i, hash := range hashes { for i, hash := range hashes {
found := false found := false
for _, msg := range result.Messages() { for _, msgHash := range result {
if bytes.Equal(msg.GetMessageHash(), hash.Bytes()) { if bytes.Equal(msgHash.Bytes(), hash.Bytes()) {
found = true found = true
break break
} }

42
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_sender.go generated vendored
View File

@ -6,9 +6,9 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/common" "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"
"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"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/time/rate" "golang.org/x/time/rate"
) )
@ -36,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 { type MessageSender struct {
publishMethod PublishMethod publishMethod PublishMethod
lightPush *lightpush.WakuLightPush publisher Publisher
relay *relay.WakuRelay
messageSentCheck ISentCheck messageSentCheck ISentCheck
rateLimiter *PublishRateLimiter rateLimiter *PublishRateLimiter
logger *zap.Logger logger *zap.Logger
@ -64,14 +74,13 @@ func (r *Request) WithPublishMethod(publishMethod PublishMethod) *Request {
return r 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 { if publishMethod == UnknownMethod {
return nil, errors.New("publish method is required") return nil, errors.New("publish method is required")
} }
return &MessageSender{ return &MessageSender{
publishMethod: publishMethod, publishMethod: publishMethod,
lightPush: lightPush, publisher: publisher,
relay: relay,
rateLimiter: NewPublishRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst), rateLimiter: NewPublishRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst),
logger: logger, logger: logger,
}, nil }, nil
@ -108,26 +117,23 @@ func (ms *MessageSender) Send(req *Request) error {
switch publishMethod { switch publishMethod {
case LightPush: case LightPush:
if ms.lightPush == nil {
return errors.New("lightpush is not available")
}
logger.Info("publishing message via lightpush") logger.Info("publishing message via lightpush")
_, err := ms.lightPush.Publish( _, err := ms.publisher.LightpushPublish(
req.ctx, req.ctx,
req.envelope.Message(), req.envelope.Message(),
lightpush.WithPubSubTopic(req.envelope.PubsubTopic()), req.envelope.PubsubTopic(),
lightpush.WithMaxPeers(DefaultPeersToPublishForLightpush), DefaultPeersToPublishForLightpush,
) )
if err != nil { if err != nil {
return err return err
} }
case Relay: case Relay:
if ms.relay == nil { peers, err := ms.publisher.RelayListPeers(req.envelope.PubsubTopic())
return errors.New("relay is not available") if err != nil {
return err
} }
peerCnt := len(ms.relay.PubSub().ListPeers(req.envelope.PubsubTopic())) logger.Info("publishing message via relay", zap.Int("peerCnt", len(peers)))
logger.Info("publishing message via relay", zap.Int("peerCnt", peerCnt)) _, err = ms.publisher.RelayPublish(req.ctx, req.envelope.Message(), req.envelope.PubsubTopic())
_, err := ms.relay.Publish(req.ctx, req.envelope.Message(), relay.WithPubSubTopic(req.envelope.PubsubTopic()))
if err != nil { if err != nil {
return err return err
} }

2
vendor/github.com/waku-org/go-waku/waku/v2/protocol/store/client.go generated vendored
View File

@ -263,7 +263,7 @@ 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) { 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") logger.Debug("sending store request")

2
vendor/modules.txt vendored
View File

@ -1040,7 +1040,7 @@ github.com/waku-org/go-discover/discover/v5wire
github.com/waku-org/go-libp2p-rendezvous github.com/waku-org/go-libp2p-rendezvous
github.com/waku-org/go-libp2p-rendezvous/db github.com/waku-org/go-libp2p-rendezvous/db
github.com/waku-org/go-libp2p-rendezvous/pb github.com/waku-org/go-libp2p-rendezvous/pb
# github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835 # github.com/waku-org/go-waku v0.8.1-0.20241021202955-3c4e40c729a0
## explicit; go 1.21 ## explicit; go 1.21
github.com/waku-org/go-waku/logging github.com/waku-org/go-waku/logging
github.com/waku-org/go-waku/tests github.com/waku-org/go-waku/tests