From d657edffc431d3dc8dd3bddc5e4104b1cdaaa5ea Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Tue, 15 Oct 2024 19:18:27 -0400 Subject: [PATCH] refactor_: decouple go-waku api from go-waku wakunode --- go.mod | 2 +- go.sum | 4 +- .../go-waku/waku/v2/api/common/result.go | 17 +++++++ .../waku/v2/api/missing/default_requestor.go | 33 ++++++++++++ .../waku/v2/api/missing/missing_messages.go | 50 +++++++++++-------- .../waku/v2/api/publish/default_publisher.go | 50 +++++++++++++++++++ .../waku/v2/api/publish/default_verifier.go | 39 +++++++++++++++ .../waku/v2/api/publish/message_check.go | 26 +++++----- .../waku/v2/api/publish/message_sender.go | 42 +++++++++------- vendor/modules.txt | 2 +- wakuv2/waku.go | 6 +-- 11 files changed, 213 insertions(+), 58 deletions(-) create mode 100644 vendor/github.com/waku-org/go-waku/waku/v2/api/common/result.go create mode 100644 vendor/github.com/waku-org/go-waku/waku/v2/api/missing/default_requestor.go create mode 100644 vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_publisher.go create mode 100644 vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_verifier.go diff --git a/go.mod b/go.mod index e6107a661..5123db98d 100644 --- a/go.mod +++ b/go.mod @@ -95,7 +95,7 @@ require ( github.com/schollz/peerdiscovery v1.7.0 github.com/siphiuel/lc-proxy-wrapper v0.0.0-20230516150924-246507cee8c7 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.20241015194815-37f936d74705 github.com/wk8/go-ordered-map/v2 v2.1.7 github.com/yeqown/go-qrcode/v2 v2.2.1 github.com/yeqown/go-qrcode/writer/standard v1.2.1 diff --git a/go.sum b/go.sum index 4fefc6689..4dc196386 100644 --- a/go.sum +++ b/go.sum @@ -2136,8 +2136,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-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-waku v0.8.1-0.20241014185851-76275f6fb835 h1:Vp6BhXiDEilmchHy8OLMZVhugudsnvveNkAKD5nhAGk= -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.20241015194815-37f936d74705 h1:i1vIOgWIQn0jing5jxqO9rG676jPoShiTLknE/pRaWc= +github.com/waku-org/go-waku v0.8.1-0.20241015194815-37f936d74705/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/go.mod h1:1PdBdPzyTaKt3VnpAHk3zj+r9dXPFOr3IHZP9nFle6E= github.com/waku-org/go-zerokit-rln-apple v0.0.0-20230916172309-ee0ee61dde2b h1:KgZVhsLkxsj5gb/FfndSCQu6VYwALrCOgYI3poR95yE= diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/common/result.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/common/result.go new file mode 100644 index 000000000..d8ce175a9 --- /dev/null +++ b/vendor/github.com/waku-org/go-waku/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 +} diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/missing/default_requestor.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/missing/default_requestor.go new file mode 100644 index 000000000..248d61c6d --- /dev/null +++ b/vendor/github.com/waku-org/go-waku/waku/v2/api/missing/default_requestor.go @@ -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)) +} diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/missing/missing_messages.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/missing/missing_messages.go index 095e32419..1af991eb5 100644 --- a/vendor/github.com/waku-org/go-waku/waku/v2/api/missing/missing_messages.go +++ b/vendor/github.com/waku-org/go-waku/waku/v2/api/missing/missing_messages.go @@ -11,9 +11,9 @@ 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" "github.com/waku-org/go-waku/waku/v2/timesource" "github.com/waku-org/go-waku/waku/v2/utils" "go.uber.org/zap" @@ -22,6 +22,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 @@ -30,25 +31,30 @@ type MessageTracker interface { 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 type MissingMessageVerifier struct { ctx context.Context 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 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 StorenodeRequestor, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier { options = append(defaultMissingMessagesVerifierOptions, options...) params := missingMessageVerifierParams{} for _, opt := range options { @@ -56,11 +62,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 +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 count := 1 for retry && count <= m.params.maxAttemptsToRetrieveHistory { @@ -212,12 +218,16 @@ 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) { + return m.storenodeRequestor.QueryWithCriteria( + ctx, + interest.peerID, + messageFetchPageSize, + 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") if err != nil { if !errors.Is(err, context.Canceled) { @@ -243,7 +253,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 +292,10 @@ 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)) + return m.storenodeRequestor.GetMessagesByHash(queryCtx, interest.peerID, maxMsgHashesPerRequest, messageHashes) }, logger, "retrieving missing messages") if err != nil { 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 { return nil, err } diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_publisher.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_publisher.go new file mode 100644 index 000000000..4ca940cef --- /dev/null +++ b/vendor/github.com/waku-org/go-waku/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)) +} diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_verifier.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_verifier.go new file mode 100644 index 000000000..68eca0304 --- /dev/null +++ b/vendor/github.com/waku-org/go-waku/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 +} diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_check.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_check.go index 059165740..8a37e20ce 100644 --- a/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_check.go +++ b/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_check.go @@ -8,11 +8,11 @@ import ( "github.com/ethereum/go-ethereum/common" "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" @@ -31,6 +31,11 @@ type ISentCheck interface { 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 // 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 @@ -40,7 +45,7 @@ type MessageSentCheck struct { 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 @@ -52,14 +57,14 @@ type MessageSentCheck struct { } // 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{ 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, @@ -212,12 +217,7 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c 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 { @@ -228,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 } diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_sender.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_sender.go index c1e9a4ca7..c457589e7 100644 --- a/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_sender.go +++ b/vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_sender.go @@ -6,9 +6,9 @@ import ( "time" "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" ) @@ -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 { publishMethod PublishMethod - lightPush *lightpush.WakuLightPush - relay *relay.WakuRelay + publisher Publisher messageSentCheck ISentCheck rateLimiter *PublishRateLimiter logger *zap.Logger @@ -64,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 @@ -108,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 } diff --git a/vendor/modules.txt b/vendor/modules.txt index 4af35ef32..7d15bf851 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1015,7 +1015,7 @@ github.com/waku-org/go-discover/discover/v5wire github.com/waku-org/go-libp2p-rendezvous github.com/waku-org/go-libp2p-rendezvous/db 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.20241015194815-37f936d74705 ## explicit; go 1.21 github.com/waku-org/go-waku/logging github.com/waku-org/go-waku/tests diff --git a/wakuv2/waku.go b/wakuv2/waku.go index 3e2319752..1869f3a42 100644 --- a/wakuv2/waku.go +++ b/wakuv2/waku.go @@ -1122,7 +1122,7 @@ func (w *Waku) Start() error { if w.cfg.EnableMissingMessageVerification { w.missingMsgVerifier = missing.NewMissingMessageVerifier( - w.node.Store(), + missing.NewDefaultStorenodeRequestor(w.node.Store()), w, w.node.Timesource(), w.logger) @@ -1267,7 +1267,7 @@ func (w *Waku) startMessageSender() error { publishMethod = publish.LightPush } - sender, err := publish.NewMessageSender(publishMethod, w.node.Lightpush(), w.node.Relay(), w.logger) + sender, err := publish.NewMessageSender(publishMethod, publish.NewDefaultPublisher(w.node.Lightpush(), w.node.Relay()), w.logger) if err != nil { w.logger.Error("failed to create message sender", zap.Error(err)) return err @@ -1276,7 +1276,7 @@ func (w *Waku) startMessageSender() error { if w.cfg.EnableStoreConfirmationForMessagesSent { msgStoredChan := make(chan gethcommon.Hash, 1000) msgExpiredChan := make(chan gethcommon.Hash, 1000) - messageSentCheck := publish.NewMessageSentCheck(w.ctx, w.node.Store(), w.StorenodeCycle, w.node.Timesource(), msgStoredChan, msgExpiredChan, w.logger) + messageSentCheck := publish.NewMessageSentCheck(w.ctx, publish.NewDefaultStorenodeMessageVerifier(w.node.Store()), w.StorenodeCycle, w.node.Timesource(), msgStoredChan, msgExpiredChan, w.logger) sender.WithMessageSentCheck(messageSentCheck) w.wg.Add(1)