mirror of
https://github.com/logos-messaging/logos-messaging-go.git
synced 2026-01-11 10:23:08 +00:00
feat: filter rate limit
Also includes refactor to have a rate limiter per peer with ttl
This commit is contained in:
Richard Ramos
parent
96702e278b
commit
5403100d91
@ -26,7 +26,7 @@ pkgs.buildGo121Module {
|
||||
'' else "";
|
||||
|
||||
# FIXME: This needs to be manually changed when updating modules.
|
||||
vendorHash = "sha256-TrKlv3UHhFl+1HviEYFTmOpF+UiVdL6h53IkJXBFsRo=";
|
||||
vendorHash = "sha256-yQ3anfZ/PU0M0KHiXqA9Ri8zFkg1nTYIk43jmcdGZYU=";
|
||||
|
||||
# Fix for 'nix run' trying to execute 'go-waku'.
|
||||
meta = { mainProgram = "waku"; };
|
||||
|
||||
@ -62,6 +62,7 @@ require (
|
||||
github.com/ipfs/go-log/v2 v2.5.1 // indirect
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
|
||||
github.com/klauspost/compress v1.17.9 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
|
||||
github.com/koron/go-ssdp v0.0.4 // indirect
|
||||
|
||||
@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
|
||||
github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
|
||||
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
|
||||
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
|
||||
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
|
||||
|
||||
@ -62,6 +62,7 @@ require (
|
||||
github.com/ipfs/go-log/v2 v2.5.1 // indirect
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
|
||||
github.com/klauspost/compress v1.17.9 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
|
||||
github.com/koron/go-ssdp v0.0.4 // indirect
|
||||
|
||||
@ -337,6 +337,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
|
||||
github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
|
||||
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
|
||||
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
|
||||
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
|
||||
|
||||
@ -70,6 +70,7 @@ require (
|
||||
github.com/ipfs/go-cid v0.4.1 // indirect
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
|
||||
github.com/klauspost/compress v1.17.9 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
|
||||
github.com/koron/go-ssdp v0.0.4 // indirect
|
||||
|
||||
@ -349,6 +349,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
|
||||
github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
|
||||
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
|
||||
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
|
||||
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
|
||||
|
||||
@ -57,6 +57,7 @@ require (
|
||||
github.com/ipfs/go-cid v0.4.1 // indirect
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
|
||||
github.com/klauspost/compress v1.17.9 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
|
||||
github.com/koron/go-ssdp v0.0.4 // indirect
|
||||
|
||||
@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
|
||||
github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
|
||||
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
|
||||
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
|
||||
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
|
||||
|
||||
@ -60,6 +60,7 @@ require (
|
||||
github.com/ipfs/go-cid v0.4.1 // indirect
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
|
||||
github.com/klauspost/compress v1.17.9 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
|
||||
github.com/koron/go-ssdp v0.0.4 // indirect
|
||||
|
||||
@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
|
||||
github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
|
||||
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
|
||||
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
|
||||
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
|
||||
|
||||
@ -58,6 +58,7 @@ require (
|
||||
github.com/ipfs/go-log/v2 v2.5.1 // indirect
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 // indirect
|
||||
github.com/klauspost/compress v1.17.9 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
|
||||
github.com/koron/go-ssdp v0.0.4 // indirect
|
||||
|
||||
@ -335,6 +335,8 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
|
||||
github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
|
||||
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
|
||||
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
|
||||
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
|
||||
|
||||
1
go.mod
1
go.mod
@ -40,6 +40,7 @@ require (
|
||||
github.com/dustin/go-humanize v1.0.1
|
||||
github.com/go-chi/chi/v5 v5.0.0
|
||||
github.com/jackc/pgx/v5 v5.4.1
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0
|
||||
github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0
|
||||
github.com/waku-org/go-noise v0.0.4
|
||||
github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59
|
||||
|
||||
2
go.sum
2
go.sum
@ -949,6 +949,8 @@ github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0
|
||||
github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
|
||||
github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e/go.mod h1:G1CVv03EnqU1wYL2dFwXxW2An0az9JTl/ZsqXQeBlkU=
|
||||
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
|
||||
github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
|
||||
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
|
||||
github.com/jinzhu/now v1.1.1/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
|
||||
|
||||
@ -85,7 +85,7 @@ type WakuNodeParameters struct {
|
||||
enableRelay bool
|
||||
enableFilterLightNode bool
|
||||
enableFilterFullNode bool
|
||||
filterOpts []filter.Option
|
||||
filterOpts []filter.FullNodeOption
|
||||
pubsubOpts []pubsub.Option
|
||||
lightpushOpts []lightpush.Option
|
||||
|
||||
@ -471,7 +471,7 @@ func WithWakuFilterLightNode() WakuNodeOption {
|
||||
|
||||
// WithWakuFilterFullNode enables the Waku Filter V2 protocol full node functionality.
|
||||
// This WakuNodeOption accepts a list of WakuFilter options to setup the protocol
|
||||
func WithWakuFilterFullNode(filterOpts ...filter.Option) WakuNodeOption {
|
||||
func WithWakuFilterFullNode(filterOpts ...filter.FullNodeOption) WakuNodeOption {
|
||||
return func(params *WakuNodeParameters) error {
|
||||
params.enableFilterFullNode = true
|
||||
params.filterOpts = filterOpts
|
||||
|
||||
@ -54,6 +54,7 @@ type WakuFilterLightNode struct {
|
||||
log *zap.Logger
|
||||
subscriptions *subscription.SubscriptionsMap
|
||||
pm *peermanager.PeerManager
|
||||
limiter *utils.RateLimiter
|
||||
peerPingInterval time.Duration
|
||||
}
|
||||
|
||||
@ -89,6 +90,7 @@ func NewWakuFilterLightNode(
|
||||
onlineChecker onlinechecker.OnlineChecker,
|
||||
reg prometheus.Registerer,
|
||||
log *zap.Logger,
|
||||
opts ...LightNodeOption,
|
||||
) *WakuFilterLightNode {
|
||||
wf := new(WakuFilterLightNode)
|
||||
wf.log = log.Named("filterv2-lightnode")
|
||||
@ -99,6 +101,14 @@ func NewWakuFilterLightNode(
|
||||
wf.CommonService = service.NewCommonService()
|
||||
wf.metrics = newMetrics(reg)
|
||||
wf.peerPingInterval = 1 * time.Minute
|
||||
|
||||
params := &FilterLightNodeParameters{}
|
||||
opts = append(DefaultLightNodeOptions(), opts...)
|
||||
for _, opt := range opts {
|
||||
opt(params)
|
||||
}
|
||||
wf.limiter = utils.NewRateLimiter(params.limitR, params.limitB)
|
||||
|
||||
return wf
|
||||
}
|
||||
|
||||
@ -155,6 +165,11 @@ func (wf *WakuFilterLightNode) onRequest(ctx context.Context) func(network.Strea
|
||||
|
||||
logger := wf.log.With(logging.HostID("peerID", peerID))
|
||||
|
||||
if !wf.limiter.Allow(peerID) {
|
||||
wf.metrics.RecordError(rateLimitFailure)
|
||||
return
|
||||
}
|
||||
|
||||
if !wf.subscriptions.IsSubscribedTo(peerID) {
|
||||
logger.Warn("received message push from unknown peer", logging.HostID("peerID", peerID))
|
||||
wf.metrics.RecordError(unknownPeerMessagePush)
|
||||
@ -287,7 +302,7 @@ func (wf *WakuFilterLightNode) request(ctx context.Context, requestID []byte,
|
||||
|
||||
}
|
||||
|
||||
if filterSubscribeResponse.RequestId != request.RequestId {
|
||||
if filterSubscribeResponse.RequestId != "N/A" && filterSubscribeResponse.RequestId != request.RequestId {
|
||||
wf.log.Error("requestID mismatch", zap.String("expected", request.RequestId), zap.String("received", filterSubscribeResponse.RequestId))
|
||||
wf.metrics.RecordError(requestIDMismatch)
|
||||
err := NewFilterError(300, "request_id_mismatch")
|
||||
|
||||
@ -96,6 +96,7 @@ var (
|
||||
peerNotFoundFailure metricsErrCategory = "peer_not_found_failure"
|
||||
writeResponseFailure metricsErrCategory = "write_response_failure"
|
||||
pushTimeoutFailure metricsErrCategory = "push_timeout_failure"
|
||||
rateLimitFailure metricsErrCategory = "ratelimit_failure"
|
||||
)
|
||||
|
||||
// RecordError increases the counter for different error types
|
||||
|
||||
@ -11,6 +11,7 @@ import (
|
||||
"github.com/waku-org/go-waku/waku/v2/peermanager"
|
||||
"github.com/waku-org/go-waku/waku/v2/protocol"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
func (old *FilterSubscribeParameters) Copy() *FilterSubscribeParameters {
|
||||
@ -53,19 +54,41 @@ type (
|
||||
wg *sync.WaitGroup
|
||||
}
|
||||
|
||||
FilterParameters struct {
|
||||
FilterFullNodeParameters struct {
|
||||
Timeout time.Duration
|
||||
MaxSubscribers int
|
||||
pm *peermanager.PeerManager
|
||||
limitR rate.Limit
|
||||
limitB int
|
||||
}
|
||||
|
||||
Option func(*FilterParameters)
|
||||
FullNodeOption func(*FilterFullNodeParameters)
|
||||
|
||||
FilterLightNodeParameters struct {
|
||||
limitR rate.Limit
|
||||
limitB int
|
||||
}
|
||||
|
||||
LightNodeOption func(*FilterLightNodeParameters)
|
||||
|
||||
FilterSubscribeOption func(*FilterSubscribeParameters) error
|
||||
)
|
||||
|
||||
func WithTimeout(timeout time.Duration) Option {
|
||||
return func(params *FilterParameters) {
|
||||
func WithLightNodeRateLimiter(r rate.Limit, b int) LightNodeOption {
|
||||
return func(params *FilterLightNodeParameters) {
|
||||
params.limitR = r
|
||||
params.limitB = b
|
||||
}
|
||||
}
|
||||
|
||||
func DefaultLightNodeOptions() []LightNodeOption {
|
||||
return []LightNodeOption{
|
||||
WithLightNodeRateLimiter(rate.Inf, 0),
|
||||
}
|
||||
}
|
||||
|
||||
func WithTimeout(timeout time.Duration) FullNodeOption {
|
||||
return func(params *FilterFullNodeParameters) {
|
||||
params.Timeout = timeout
|
||||
}
|
||||
}
|
||||
@ -190,21 +213,29 @@ func DefaultUnsubscribeOptions() []FilterSubscribeOption {
|
||||
}
|
||||
}
|
||||
|
||||
func WithMaxSubscribers(maxSubscribers int) Option {
|
||||
return func(params *FilterParameters) {
|
||||
func WithMaxSubscribers(maxSubscribers int) FullNodeOption {
|
||||
return func(params *FilterFullNodeParameters) {
|
||||
params.MaxSubscribers = maxSubscribers
|
||||
}
|
||||
}
|
||||
|
||||
func WithPeerManager(pm *peermanager.PeerManager) Option {
|
||||
return func(params *FilterParameters) {
|
||||
func WithPeerManager(pm *peermanager.PeerManager) FullNodeOption {
|
||||
return func(params *FilterFullNodeParameters) {
|
||||
params.pm = pm
|
||||
}
|
||||
}
|
||||
|
||||
func DefaultOptions() []Option {
|
||||
return []Option{
|
||||
WithTimeout(DefaultIdleSubscriptionTimeout),
|
||||
WithMaxSubscribers(DefaultMaxSubscribers),
|
||||
func WithFullNodeRateLimiter(r rate.Limit, b int) FullNodeOption {
|
||||
return func(params *FilterFullNodeParameters) {
|
||||
params.limitR = r
|
||||
params.limitB = b
|
||||
}
|
||||
}
|
||||
|
||||
func DefaultFullNodeOptions() []FullNodeOption {
|
||||
return []FullNodeOption{
|
||||
WithTimeout(DefaultIdleSubscriptionTimeout),
|
||||
WithMaxSubscribers(DefaultMaxSubscribers),
|
||||
WithFullNodeRateLimiter(rate.Inf, 0),
|
||||
}
|
||||
}
|
||||
|
||||
@ -37,25 +37,26 @@ type (
|
||||
metrics Metrics
|
||||
log *zap.Logger
|
||||
*service.CommonService
|
||||
subscriptions *SubscribersMap
|
||||
pm *peermanager.PeerManager
|
||||
|
||||
subscriptions *SubscribersMap
|
||||
pm *peermanager.PeerManager
|
||||
limiter *utils.RateLimiter
|
||||
maxSubscriptions int
|
||||
}
|
||||
)
|
||||
|
||||
// NewWakuFilterFullNode returns a new instance of Waku Filter struct setup according to the chosen parameter and options
|
||||
func NewWakuFilterFullNode(timesource timesource.Timesource, reg prometheus.Registerer, log *zap.Logger, opts ...Option) *WakuFilterFullNode {
|
||||
func NewWakuFilterFullNode(timesource timesource.Timesource, reg prometheus.Registerer, log *zap.Logger, opts ...FullNodeOption) *WakuFilterFullNode {
|
||||
wf := new(WakuFilterFullNode)
|
||||
wf.log = log.Named("filterv2-fullnode")
|
||||
|
||||
params := new(FilterParameters)
|
||||
optList := DefaultOptions()
|
||||
params := new(FilterFullNodeParameters)
|
||||
optList := DefaultFullNodeOptions()
|
||||
optList = append(optList, opts...)
|
||||
for _, opt := range optList {
|
||||
opt(params)
|
||||
}
|
||||
|
||||
wf.limiter = utils.NewRateLimiter(params.limitR, params.limitB)
|
||||
wf.CommonService = service.NewCommonService()
|
||||
wf.metrics = newMetrics(reg)
|
||||
wf.subscriptions = NewSubscribersMap(params.Timeout)
|
||||
@ -93,7 +94,14 @@ func (wf *WakuFilterFullNode) start(sub *relay.Subscription) error {
|
||||
|
||||
func (wf *WakuFilterFullNode) onRequest(ctx context.Context) func(network.Stream) {
|
||||
return func(stream network.Stream) {
|
||||
logger := wf.log.With(logging.HostID("peer", stream.Conn().RemotePeer()))
|
||||
peerID := stream.Conn().RemotePeer()
|
||||
logger := wf.log.With(logging.HostID("peer", peerID))
|
||||
|
||||
if !wf.limiter.Allow(peerID) {
|
||||
wf.metrics.RecordError(rateLimitFailure)
|
||||
wf.reply(ctx, stream, &pb.FilterSubscribeRequest{RequestId: "N/A"}, http.StatusTooManyRequests, "filter request rejected due rate limit exceeded")
|
||||
return
|
||||
}
|
||||
|
||||
reader := pbio.NewDelimitedReader(stream, math.MaxInt32)
|
||||
|
||||
|
||||
@ -24,7 +24,6 @@ import (
|
||||
"github.com/waku-org/go-waku/waku/v2/protocol/relay"
|
||||
"github.com/waku-org/go-waku/waku/v2/utils"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
// LightPushID_v20beta1 is the current Waku LightPush protocol identifier
|
||||
@ -40,7 +39,7 @@ var (
|
||||
type WakuLightPush struct {
|
||||
h host.Host
|
||||
relay *relay.WakuRelay
|
||||
limiter *rate.Limiter
|
||||
limiter *utils.RateLimiter
|
||||
cancel context.CancelFunc
|
||||
pm *peermanager.PeerManager
|
||||
metrics Metrics
|
||||
@ -59,11 +58,12 @@ func NewWakuLightPush(relay *relay.WakuRelay, pm *peermanager.PeerManager, reg p
|
||||
wakuLP.metrics = newMetrics(reg)
|
||||
|
||||
params := &LightpushParameters{}
|
||||
opts = append(DefaultOptions(), opts...)
|
||||
for _, opt := range opts {
|
||||
opt(params)
|
||||
}
|
||||
|
||||
wakuLP.limiter = params.limiter
|
||||
wakuLP.limiter = utils.NewRateLimiter(params.limitR, params.limitB)
|
||||
|
||||
return wakuLP
|
||||
}
|
||||
@ -106,7 +106,7 @@ func (wakuLP *WakuLightPush) onRequest(ctx context.Context) func(network.Stream)
|
||||
Response: &pb.PushResponse{},
|
||||
}
|
||||
|
||||
if wakuLP.limiter != nil && !wakuLP.limiter.Allow() {
|
||||
if !wakuLP.limiter.Allow(stream.Conn().RemotePeer()) {
|
||||
wakuLP.metrics.RecordError(rateLimitFailure)
|
||||
responseMsg := "exceeds the rate limit"
|
||||
responsePushRPC.Response.Info = &responseMsg
|
||||
@ -257,7 +257,7 @@ func (wakuLP *WakuLightPush) handleOpts(ctx context.Context, message *wpb.WakuMe
|
||||
params.pm = wakuLP.pm
|
||||
var err error
|
||||
|
||||
optList := append(DefaultOptions(wakuLP.h), opts...)
|
||||
optList := append(DefaultRequestOptions(wakuLP.h), opts...)
|
||||
for _, opt := range optList {
|
||||
err := opt(params)
|
||||
if err != nil {
|
||||
|
||||
@ -14,7 +14,8 @@ import (
|
||||
)
|
||||
|
||||
type LightpushParameters struct {
|
||||
limiter *rate.Limiter
|
||||
limitR rate.Limit
|
||||
limitB int
|
||||
}
|
||||
|
||||
type Option func(*LightpushParameters)
|
||||
@ -22,7 +23,14 @@ type Option func(*LightpushParameters)
|
||||
// WithRateLimiter is an option used to specify a rate limiter for requests received in lightpush protocol
|
||||
func WithRateLimiter(r rate.Limit, b int) Option {
|
||||
return func(params *LightpushParameters) {
|
||||
params.limiter = rate.NewLimiter(r, b)
|
||||
params.limitR = r
|
||||
params.limitB = b
|
||||
}
|
||||
}
|
||||
|
||||
func DefaultOptions() []Option {
|
||||
return []Option{
|
||||
WithRateLimiter(rate.Inf, 0),
|
||||
}
|
||||
}
|
||||
|
||||
@ -130,8 +138,8 @@ func WithAutomaticRequestID() RequestOption {
|
||||
}
|
||||
}
|
||||
|
||||
// DefaultOptions are the default options to be used when using the lightpush protocol
|
||||
func DefaultOptions(host host.Host) []RequestOption {
|
||||
// DefaultRequestOptions are the default options to be used when using the lightpush protocol
|
||||
func DefaultRequestOptions(host host.Host) []RequestOption {
|
||||
return []RequestOption{
|
||||
WithAutomaticPeerSelection(),
|
||||
WithMaxPeers(1), //keeping default as 2 for status use-case
|
||||
|
||||
@ -26,7 +26,7 @@ func (wakuPX *WakuPeerExchange) Request(ctx context.Context, numPeers int, opts
|
||||
params.log = wakuPX.log
|
||||
params.pm = wakuPX.pm
|
||||
|
||||
optList := DefaultOptions(wakuPX.h)
|
||||
optList := DefaultRequestOptions(wakuPX.h)
|
||||
optList = append(optList, opts...)
|
||||
for _, opt := range optList {
|
||||
err := opt(params)
|
||||
|
||||
@ -23,7 +23,6 @@ import (
|
||||
"github.com/waku-org/go-waku/waku/v2/service"
|
||||
"github.com/waku-org/go-waku/waku/v2/utils"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
// PeerExchangeID_v20alpha1 is the current Waku Peer Exchange protocol identifier
|
||||
@ -51,7 +50,7 @@ type WakuPeerExchange struct {
|
||||
|
||||
peerConnector PeerConnector
|
||||
enrCache *enrCache
|
||||
limiter *rate.Limiter
|
||||
limiter *utils.RateLimiter
|
||||
}
|
||||
|
||||
// NewWakuPeerExchange returns a new instance of WakuPeerExchange struct
|
||||
@ -68,11 +67,12 @@ func NewWakuPeerExchange(disc *discv5.DiscoveryV5, clusterID uint16, peerConnect
|
||||
wakuPX.CommonService = service.NewCommonService()
|
||||
|
||||
params := &PeerExchangeParameters{}
|
||||
opts = append(DefaultOptions(), opts...)
|
||||
for _, opt := range opts {
|
||||
opt(params)
|
||||
}
|
||||
|
||||
wakuPX.limiter = params.limiter
|
||||
wakuPX.limiter = utils.NewRateLimiter(params.limiterR, params.limiterB)
|
||||
return wakuPX, nil
|
||||
}
|
||||
|
||||
@ -97,9 +97,10 @@ func (wakuPX *WakuPeerExchange) start() error {
|
||||
|
||||
func (wakuPX *WakuPeerExchange) onRequest() func(network.Stream) {
|
||||
return func(stream network.Stream) {
|
||||
logger := wakuPX.log.With(logging.HostID("peer", stream.Conn().RemotePeer()))
|
||||
peerID := stream.Conn().RemotePeer()
|
||||
logger := wakuPX.log.With(logging.HostID("peer", peerID))
|
||||
|
||||
if wakuPX.limiter != nil && !wakuPX.limiter.Allow() {
|
||||
if wakuPX.limiter != nil && !wakuPX.limiter.Allow(peerID) {
|
||||
wakuPX.metrics.RecordError(rateLimitFailure)
|
||||
wakuPX.log.Info("exceeds the rate limit")
|
||||
// TODO: peer exchange protocol should contain an err field
|
||||
|
||||
@ -12,7 +12,8 @@ import (
|
||||
)
|
||||
|
||||
type PeerExchangeParameters struct {
|
||||
limiter *rate.Limiter
|
||||
limiterR rate.Limit
|
||||
limiterB int
|
||||
}
|
||||
|
||||
type Option func(*PeerExchangeParameters)
|
||||
@ -20,7 +21,14 @@ type Option func(*PeerExchangeParameters)
|
||||
// WithRateLimiter is an option used to specify a rate limiter for requests received in lightpush protocol
|
||||
func WithRateLimiter(r rate.Limit, b int) Option {
|
||||
return func(params *PeerExchangeParameters) {
|
||||
params.limiter = rate.NewLimiter(r, b)
|
||||
params.limiterR = r
|
||||
params.limiterB = b
|
||||
}
|
||||
}
|
||||
|
||||
func DefaultOptions() []Option {
|
||||
return []Option{
|
||||
WithRateLimiter(rate.Inf, 0),
|
||||
}
|
||||
}
|
||||
|
||||
@ -87,8 +95,8 @@ func WithFastestPeerSelection(fromThesePeers ...peer.ID) RequestOption {
|
||||
}
|
||||
}
|
||||
|
||||
// DefaultOptions are the default options to be used when using the lightpush protocol
|
||||
func DefaultOptions(host host.Host) []RequestOption {
|
||||
// DefaultRequestOptions are the default options to be used when using the lightpush protocol
|
||||
func DefaultRequestOptions(host host.Host) []RequestOption {
|
||||
return []RequestOption{
|
||||
WithAutomaticPeerSelection(),
|
||||
}
|
||||
|
||||
@ -248,7 +248,7 @@ func TestPeerExchangeOptions(t *testing.T) {
|
||||
params.log = px1.log
|
||||
params.pm = px1.pm
|
||||
|
||||
optList := DefaultOptions(px1.h)
|
||||
optList := DefaultRequestOptions(px1.h)
|
||||
optList = append(optList, WithPeerAddr(host1.Addrs()[0]))
|
||||
for _, opt := range optList {
|
||||
err := opt(params)
|
||||
@ -258,7 +258,7 @@ func TestPeerExchangeOptions(t *testing.T) {
|
||||
require.Equal(t, host1.Addrs()[0], params.peerAddr)
|
||||
|
||||
// Test WithFastestPeerSelection()
|
||||
optList = DefaultOptions(px1.h)
|
||||
optList = DefaultRequestOptions(px1.h)
|
||||
optList = append(optList, WithFastestPeerSelection(host1.ID()))
|
||||
for _, opt := range optList {
|
||||
err := opt(params)
|
||||
|
||||
72
waku/v2/utils/limiter.go
Normal file
72
waku/v2/utils/limiter.go
Normal file
@ -0,0 +1,72 @@
|
||||
package utils
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/jellydator/ttlcache/v3"
|
||||
"github.com/libp2p/go-libp2p/core/peer"
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
type RateLimiter struct {
|
||||
sync.Mutex
|
||||
limiters *ttlcache.Cache[peer.ID, *rate.Limiter]
|
||||
r rate.Limit
|
||||
b int
|
||||
}
|
||||
|
||||
func NewRateLimiter(r rate.Limit, b int) *RateLimiter {
|
||||
return &RateLimiter{
|
||||
r: r,
|
||||
b: b,
|
||||
limiters: ttlcache.New[peer.ID, *rate.Limiter](
|
||||
ttlcache.WithTTL[peer.ID, *rate.Limiter](30 * time.Minute),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *RateLimiter) Start(ctx context.Context) {
|
||||
go func() {
|
||||
t := time.NewTicker(time.Hour)
|
||||
defer t.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
r.Lock()
|
||||
r.limiters.DeleteExpired()
|
||||
r.Unlock()
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (r *RateLimiter) getOrCreate(peerID peer.ID) *rate.Limiter {
|
||||
r.Lock()
|
||||
defer r.Unlock()
|
||||
|
||||
var limiter *rate.Limiter
|
||||
if !r.limiters.Has(peerID) {
|
||||
limiter = rate.NewLimiter(r.r, r.b)
|
||||
r.limiters.Set(peerID, limiter, ttlcache.DefaultTTL)
|
||||
} else {
|
||||
v := r.limiters.Get(peerID)
|
||||
limiter = v.Value()
|
||||
}
|
||||
return limiter
|
||||
}
|
||||
|
||||
func (r *RateLimiter) Allow(peerID peer.ID) bool {
|
||||
return r.getOrCreate(peerID).Allow()
|
||||
}
|
||||
|
||||
func (r *RateLimiter) Wait(ctx context.Context, peerID peer.ID) error {
|
||||
r.Lock()
|
||||
defer r.Unlock()
|
||||
|
||||
return r.getOrCreate(peerID).Wait(ctx)
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user