From 188220c00aaebb5fbb34e63c29afcf8f544878ee Mon Sep 17 00:00:00 2001 From: "Hayim.Shaul@ibm.com" Date: Tue, 18 Aug 2026 08:55:51 +0000 Subject: [PATCH] implement an example rate limiter Signed-off-by: Hayim.Shaul@ibm.com --- docs/README.md | 2 +- docs/configuration.md | 27 ++ docs/security/selector_resource_limits.md | 205 ++++++++--- token/selector.go | 12 +- token/services/selector/config/driver.go | 64 ++++ token/services/selector/config/driver_test.go | 99 ++++++ .../services/selector/ratelimit/decorator.go | 95 +++++ .../selector/ratelimit/decorator_test.go | 250 +++++++++++++ token/services/selector/ratelimit/limiter.go | 328 +++++++++++++++++ .../selector/ratelimit/limiter_test.go | 329 ++++++++++++++++++ token/services/selector/ratelimit/options.go | 86 +++++ .../selector/ratelimit/options_test.go | 90 +++++ .../selector/sherdlock/ratelimit_test.go | 180 ++++++++++ token/services/selector/sherdlock/service.go | 26 +- .../selector/simple/ratelimit_test.go | 111 ++++++ token/services/selector/simple/service.go | 42 ++- 16 files changed, 1875 insertions(+), 71 deletions(-) create mode 100644 token/services/selector/ratelimit/decorator.go create mode 100644 token/services/selector/ratelimit/decorator_test.go create mode 100644 token/services/selector/ratelimit/limiter.go create mode 100644 token/services/selector/ratelimit/limiter_test.go create mode 100644 token/services/selector/ratelimit/options.go create mode 100644 token/services/selector/ratelimit/options_test.go create mode 100644 token/services/selector/sherdlock/ratelimit_test.go create mode 100644 token/services/selector/simple/ratelimit_test.go diff --git a/docs/README.md b/docs/README.md index 59fe04938b..31e4c515cd 100644 --- a/docs/README.md +++ b/docs/README.md @@ -16,7 +16,7 @@ Welcome to Panurus documentation. ## Security * [**HTLC Deadlines and Clock Synchronisation**](security/htlc_deadline_clock_assumptions.md): The clock-synchronisation assumption that the HTLC claim/reclaim deadline rests on, and the deadline margin it requires of a deployment. -* [**Selector Resource Limits**](security/selector_resource_limits.md): How to throttle token selection by supplying a custom `Locker`. +* [**Selector Resource Limits**](security/selector_resource_limits.md): How to throttle token selection, either with the opt-in built-in per-wallet rate limiter or by supplying your own limiter or `Locker`. ## Command-Line Tools diff --git a/docs/configuration.md b/docs/configuration.md index 55537dbc06..3a6a413773 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -39,6 +39,24 @@ token: # fetcherCacheMaxQueries is the number of queries after which a soft refresh (non-blocking background update) is triggered. # This helps keep the cache fresh without blocking queries. If not specified or set to 0, defaults to 5 queries. fetcherCacheMaxQueries: 5 + # Built-in per-wallet rate limiter for token selection (both drivers). + # It is disabled by default: without these keys, selection requests are not metered. + # One selection request (a Selector.Select call) costs one unit, no matter how many tokens it + # locks or how often it retries internally. Unlocking tokens is never throttled. + # A throttled request fails fast with an error wrapping token.SelectorRateLimited. + # See docs/security/selector_resource_limits.md. + # rateLimitEnabled turns the limiter on with the default rate and burst below. + rateLimitEnabled: true + # rateLimit is the maximum number of selection requests per second a single wallet may issue. + # A positive value implies rateLimitEnabled: true. If not specified or set to 0, defaults to 100. + rateLimit: 100 + # rateLimitBurst is the maximum number of selection requests a single wallet may issue + # back-to-back. If not specified or set to 0, defaults to twice rateLimit. + rateLimitBurst: 200 + # rateLimitMaxBuckets caps the number of per-wallet buckets kept in memory. When the cap is + # reached, idle buckets are pruned first and, if that is not enough, the least recently used + # ones are dropped. If not specified or set to 0, defaults to 4096. + rateLimitMaxBuckets: 4096 # When we are interested in knowing when a transaction reaches finality, we subscribe to the Finality Listener Manager for the finality event of that transaction. # This configuration specifies the way the manager is instantiated (i.e., how it gets notified about the finality events, how often it checks). @@ -471,6 +489,15 @@ Default values: - numRetries: 3 - leaseExpiry: 3m - leaseCleanupTickPeriod: 90s +- rateLimitEnabled: false (the built-in per-wallet selection rate limiter is opt-in) +- rateLimit: 100 requests/s per wallet, when rate limiting is enabled +- rateLimitBurst: 2 × rateLimit, so 200 requests, when rate limiting is enabled +- rateLimitMaxBuckets: 4096 + +Setting a positive `rateLimit` is enough to enable the limiter; `rateLimitEnabled: true` alone +enables it with the defaults above. See +[docs/security/selector_resource_limits.md](security/selector_resource_limits.md) for what is +metered and how to plug in your own limiter instead. --- diff --git a/docs/security/selector_resource_limits.md b/docs/security/selector_resource_limits.md index b70ceffc16..f9b6685cb7 100644 --- a/docs/security/selector_resource_limits.md +++ b/docs/security/selector_resource_limits.md @@ -4,39 +4,150 @@ Token selection acquires a short-lived *lock* on each candidate token so that two concurrent transactions do not try to spend the same token. Under load, a single -wallet can drive a large number of selection/lock requests. Applications that need to -throttle this — to protect the lock store, to enforce fairness between wallets, or to -integrate with an existing quota system — can do so by supplying their own `Locker` -implementation. +wallet can drive a large number of selection/lock requests. To protect the lock store, +to enforce fairness between wallets, or to integrate with an existing quota system, +Panurus offers two ways to throttle this: + +1. A **built-in per-wallet rate limiter**, activated purely from configuration or from + code. It is a token bucket per wallet, in process, and it is **disabled by default**. +2. A **fail-fast contract**, `token.SelectorRateLimited`, plus a **wallet-id-aware lock + function**. Both selector drivers (simple and sherdlock) pass the wallet id the tokens + are being selected for into the `Locker`'s lock function, and abort the selection + immediately when a lock is denied with an error wrapping that sentinel. This is the + integration point for applications that would rather reuse the rate-limiting + infrastructure they already run (for example a Redis-backed limiter shared across + processes). + +## The built-in limiter + +Package `token/services/selector/ratelimit`. One **selection request** — one +`Selector.Select` call — costs one unit from the bucket of the wallet it selects for, +regardless of how many tokens the request ends up locking or how many times the selector +retries internally on contention. Deliberately *not* per token lock attempt: charging +there would make a large transfer cost more than a small one and would let the selector's +own contention retries drain a wallet's allowance. + +Properties worth knowing: + +- **Per wallet, per TMS.** Buckets are keyed by the wallet id *and* the TMS id, so the + same wallet id in two networks or namespaces gets two independent allowances. +- **Bounded memory.** Buckets are created on first use and pruned without any background + goroutine: idle ones are swept during ordinary access, and a hard cap + (`rateLimitMaxBuckets`) evicts the least recently used ones if the sweep is not enough. +- **Unlocking is never throttled.** Only `Select` is metered; `Unlock` and `Close` pass + through, so a throttled wallet can always clean up after itself. +- **Empty wallet ids are never throttled.** +- **Nothing is leaked on a denial.** The request is rejected before the selector runs, so + no token is locked and there is nothing to release. + +### Enabling it from configuration + +Under `token.selector` (see [../configuration.md](../configuration.md)): + +```yaml +token: + selector: + driver: sherdlock + # Enables the limiter with the defaults below. + rateLimitEnabled: true + # Selection requests per second per wallet. A positive value implies rateLimitEnabled: true. + # Defaults to 100. + rateLimit: 100 + # Back-to-back requests allowed to one wallet. Defaults to 2 × rateLimit, so 200. + rateLimitBurst: 200 + # Maximum number of per-wallet buckets held in memory. Defaults to 4096. + rateLimitMaxBuckets: 4096 +``` + +Omitting all four keys, which is the default, leaves selection unmetered. + +### Enabling it from code + +Both selector services accept `ratelimit.Option` values, which take precedence over the +configuration: + +```go +import ( + "github.com/LFDT-Panurus/panurus/token/services/selector/ratelimit" + "github.com/LFDT-Panurus/panurus/token/services/selector/sherdlock" +) + +// The built-in limiter with its defaults (100 requests/s per wallet, burst 200). +svc := sherdlock.NewService(fetcherProvider, lockStoreManager, configProvider, metricsProvider, + ratelimit.WithDefaultLimiter()) + +// Explicit rate and burst. +svc = sherdlock.NewService(fetcherProvider, lockStoreManager, configProvider, metricsProvider, + ratelimit.WithLimiter(ratelimit.New(ratelimit.Config{Rate: 20, Burst: 40}))) + +// Explicitly off, whatever the configuration says. +svc = sherdlock.NewService(fetcherProvider, lockStoreManager, configProvider, metricsProvider, + ratelimit.WithLimiter(nil)) +``` + +`simple.NewService(lockerProvider, configProvider, opts ...ratelimit.Option)` takes the +same options. + +A limiter passed with `WithLimiter` belongs to the caller: the service never stops it, not +even from `Shutdown`. That matters because `Shutdown` also runs on routine public-parameter +reloads, and resetting every wallet's bucket there would let a throttled client wash out +its debt. `BucketLimiter.Stop()` exists for callers that own a limiter and want its memory +back. -Panurus deliberately ships **no built-in rate limiter or quota**. Instead it -gives you two things: +### Supplying your own limiter -1. A **wallet-id-aware lock function**. Both selector drivers (simple and sherdlock) - pass the wallet id the tokens are being selected for into the `Locker`'s lock - function, so a custom `Locker` can apply per-wallet policies. -2. A **fail-fast contract**, `token.SelectorRateLimited`. When a `Locker` denies a - lock by returning an error that wraps this sentinel, the selector aborts the - selection immediately and returns the error to the caller instead of retrying. +`WithLimiter` accepts any implementation of: -This keeps the Panurus minimal and lets applications reuse whatever rate-limiting -infrastructure they already run (for example a Redis-backed limiter shared across -processes). +```go +// Limiter meters token selection requests per wallet. +type Limiter interface { + // Allow returns nil when a selection request for walletID within scope may proceed, + // and an error wrapping token.SelectorRateLimited when it must be denied. + // scope is the TMS id. An empty walletID is never throttled. + Allow(ctx context.Context, scope string, walletID string) error +} +``` + +This is the simplest way to plug in a shared, cluster-wide limiter (Redis, a quota table, +a sidecar) while keeping the metering point — one unit per selection request — and the +fail-fast behaviour that Panurus already implements. + +## The fail-fast contract + +`token/selector.go` defines: + +```go +// SelectorRateLimited is the contract error returned (directly or wrapped) to deny a +// selection for policy reasons such as rate limiting or quota. +var SelectorRateLimited = errors.New("selection rate limit exceeded") +``` + +When a `Locker` returns an error `e` with `errors.Is(e, token.SelectorRateLimited)`, the +selector: + +- stops iterating candidate tokens, +- releases any tokens it already locked for this request, and +- returns `e` to the caller. + +Any *other* error from the lock function keeps the existing semantics: the token is +treated as unavailable (e.g. already locked by another transaction) and selection +continues / retries as before. ## The lock function Both selector drivers route through a `Locker` whose lock function receives the -wallet id. +wallet id, so a custom `Locker` can apply per-wallet policies of its own — per token +lock rather than per selection request. **Simple selector** — `token/services/selector/simple/selector.go`: ```go type Locker interface { - // Lock locks the token id for the consumer transaction txID on behalf of walletID + // Lock locks the token id for the consumer transaction txID on behalf of owner // (ownerFilter.ID()). Return an error wrapping token.SelectorRateLimited to deny // the lock and make the selection fail fast. - Lock(ctx context.Context, id *token.ID, txID string, walletID string, reclaim bool) (string, error) - UnlockIDs(ctx context.Context, ids ...*token.ID) []*token.ID + Lock(ctx context.Context, owner string, id *token.ID, txID string, reclaim bool) (string, error) + UnlockIDs(ctx context.Context, owner string, ids ...*token.ID) []*token.ID UnlockByTxID(ctx context.Context, txID string) IsLocked(id *token.ID) bool } @@ -60,34 +171,12 @@ type TokenLockStore interface { The built-in in-memory locker and the SQL-backed `TokenLockStore` accept `walletID` but do not act on it — they apply no rate limiting or quota. -## The fail-fast contract - -`token/selector.go` defines: - -```go -// SelectorRateLimited is the contract error a Locker implementation returns (directly -// or wrapped) to deny a lock for policy reasons such as rate limiting or quota. -var SelectorRateLimited = errors.New("selection rate limit exceeded") -``` - -When your `Locker` returns an error `e` with `errors.Is(e, token.SelectorRateLimited)`, -the selector: - -- stops iterating candidate tokens, -- releases any tokens it already locked for this request, and -- returns `e` to the caller. - -Any *other* error from the lock function keeps the existing semantics: the token is -treated as unavailable (e.g. already locked by another transaction) and selection -continues / retries as before. - -## Integrating your own rate limiting +### Integrating your own rate limiting in a Locker Provide a `Locker` that wraps the SDK's default locker and enforces your policy before delegating. Below, a Redis-backed limiter throttles per wallet; the same shape works -for an in-process limiter, a quota table, etc. - -### Simple selector +for an in-process limiter, a quota table, etc. Note that this is charged **per token lock +attempt**, unlike the built-in limiter above. ```go import ( @@ -105,24 +194,21 @@ type rateLimitedLocker struct { limiter RedisLimiter // your existing infrastructure } -func (l *rateLimitedLocker) Lock(ctx context.Context, id *tokenapi.ID, txID string, walletID string, reclaim bool) (string, error) { - if !l.limiter.Allow(ctx, walletID) { - return "", errors.Wrapf(token.SelectorRateLimited, "wallet %s throttled", walletID) +func (l *rateLimitedLocker) Lock(ctx context.Context, owner string, id *tokenapi.ID, txID string, reclaim bool) (string, error) { + if !l.limiter.Allow(ctx, owner) { + return "", errors.Wrapf(token.SelectorRateLimited, "wallet %s throttled", owner) } - return l.Locker.Lock(ctx, id, txID, walletID, reclaim) + return l.Locker.Lock(ctx, owner, id, txID, reclaim) } ``` Wire it in by providing a `simple.LockerProvider` whose `New` returns your decorator instead of the default `inmemory.NewLocker`. -### Sherdlock selector - -Provide a `TokenLockStore` (via the `tokenlockdb.StoreServiceManager` used by -`sherdlock.NewService`) whose `Lock` enforces the limit before delegating to the -SQL-backed store, returning an error wrapping `token.SelectorRateLimited` when a wallet -is throttled. +For the sherdlock selector, provide a `TokenLockStore` (via the +`tokenlockdb.StoreServiceManager` used by `sherdlock.NewService`) whose `Lock` enforces the +limit before delegating to the SQL-backed store. ## Handling the error @@ -136,9 +222,14 @@ if errors.Is(err, token.SelectorRateLimited) { } ``` +The built-in limiter's error message states how long to wait before the wallet has a +request available again. + ## Notes - Passing an empty `walletID` is valid; a `Locker` that keys its policy on wallet id - should treat empty as "no throttling" (the default lockers ignore it entirely). -- Because the policy lives in your `Locker`, its scope (per process vs shared across a - cluster), persistence, and lifecycle are entirely under your control. + should treat empty as "no throttling" (the default lockers ignore it entirely, and so + does the built-in limiter). +- The built-in limiter is per process. If a wallet's traffic is spread over several nodes, + each node enforces its own allowance; supply a shared `Limiter` if you need a + cluster-wide budget. diff --git a/token/selector.go b/token/selector.go index f4a7b46b62..1e5ec858f6 100644 --- a/token/selector.go +++ b/token/selector.go @@ -25,13 +25,15 @@ var ( // SelectorSufficientFundsButConcurrencyIssue is returned when funds are sufficient to cover the request, but // concurrency issues does not make some of the selected tokens available. SelectorSufficientFundsButConcurrencyIssue = errors.New("sufficient funds but concurrency issue") - // SelectorRateLimited is the contract error a Locker implementation returns (directly - // or wrapped) to deny a lock for policy reasons such as rate limiting or quota. + // SelectorRateLimited is the contract error returned (directly or wrapped) to deny a + // selection for policy reasons such as rate limiting or quota. // Both the simple and sherdlock selectors detect it via errors.Is and abort the // selection immediately, returning the error to the caller instead of retrying. - // Panurus ships no built-in limiter: applications integrate their own - // (e.g. a Redis-backed limiter) by providing a Locker implementation that returns - // this error when a request must be throttled. + // Panurus ships an opt-in per-wallet limiter that returns it, see + // token/services/selector/ratelimit and the token.selector.rateLimit* configuration + // keys; it is disabled by default. Applications that would rather reuse their own + // infrastructure (e.g. a Redis-backed limiter) can either supply a Limiter to that + // package or return this error from a custom Locker implementation. SelectorRateLimited = errors.New("selection rate limit exceeded") ) diff --git a/token/services/selector/config/driver.go b/token/services/selector/config/driver.go index 5bff7d8553..ef6bdc6f89 100644 --- a/token/services/selector/config/driver.go +++ b/token/services/selector/config/driver.go @@ -7,9 +7,11 @@ SPDX-License-Identifier: Apache-2.0 package config import ( + "math" "time" "github.com/LFDT-Panurus/panurus/token/services/selector/driver" + "github.com/LFDT-Panurus/panurus/token/services/selector/ratelimit" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" ) @@ -22,6 +24,12 @@ const ( defaultFetcherCacheSize = 0 // 0 means use fetcher default defaultFetcherCacheRefresh = 0 // 0 means use fetcher default defaultFetcherCacheMaxQueries = 0 // 0 means use fetcher default + // defaultRateLimit is the per-wallet selection rate, in requests per second, applied when + // rate limiting is enabled without an explicit rateLimit. + defaultRateLimit = ratelimit.DefaultRate + // defaultRateLimitBurstFactor multiplies the rate to obtain the burst capacity applied + // when rate limiting is enabled without an explicit rateLimitBurst. + defaultRateLimitBurstFactor = ratelimit.DefaultBurstFactor ) //go:generate counterfeiter -o mock/config_service.go -fake-name ConfigService . configService @@ -38,6 +46,21 @@ type Config struct { FetcherCacheSize int64 `yaml:"fetcherCacheSize,omitempty"` FetcherCacheRefresh time.Duration `yaml:"fetcherCacheRefresh,omitempty"` FetcherCacheMaxQueries int `yaml:"fetcherCacheMaxQueries,omitempty"` + // RateLimitEnabled turns on the built-in per-wallet selection rate limiter with the + // default rate and burst. Rate limiting is off unless this is set or RateLimit is + // positive. + RateLimitEnabled bool `yaml:"rateLimitEnabled,omitempty"` + // RateLimit is the maximum number of selection requests per second a single wallet may + // issue. A positive value implies RateLimitEnabled. A value <= 0 falls back to the + // default rate when rate limiting is enabled some other way, and means "no limit" + // otherwise. + RateLimit float64 `yaml:"rateLimit,omitempty"` + // RateLimitBurst is the maximum number of selection requests a single wallet may issue + // back-to-back. If <= 0, it defaults to defaultRateLimitBurstFactor times the rate. + RateLimitBurst int `yaml:"rateLimitBurst,omitempty"` + // RateLimitMaxBuckets caps the number of wallet buckets the limiter keeps in memory. If + // <= 0, the limiter's own default is used. + RateLimitMaxBuckets int `yaml:"rateLimitMaxBuckets,omitempty"` } // New returns a SelectorConfig with the values from the token.selector key @@ -105,3 +128,44 @@ func (c *Config) GetFetcherCacheMaxQueries() int { // Return 0 if not set, which will trigger use of fetcher default return c.FetcherCacheMaxQueries } + +// IsRateLimitEnabled tells whether the built-in per-wallet selection rate limiter must be +// activated. It is off by default: either rateLimitEnabled is set explicitly, or a positive +// rateLimit is configured, which implies it. +func (c *Config) IsRateLimitEnabled() bool { + return c.RateLimitEnabled || c.RateLimit > 0 +} + +// GetRateLimit returns the maximum number of selection requests per second allowed to a single +// wallet, or 0 when rate limiting is disabled. When enabled without an explicit positive +// rateLimit, it returns defaultRateLimit. +func (c *Config) GetRateLimit() float64 { + if !c.IsRateLimitEnabled() { + return 0 + } + if c.RateLimit > 0 { + return c.RateLimit + } + + return defaultRateLimit +} + +// GetRateLimitBurst returns the burst capacity of a wallet bucket, or 0 when rate limiting is +// disabled. When enabled without an explicit positive rateLimitBurst, it returns +// defaultRateLimitBurstFactor times the rate, rounded up, and never less than one request. +func (c *Config) GetRateLimitBurst() int { + if !c.IsRateLimitEnabled() { + return 0 + } + if c.RateLimitBurst > 0 { + return c.RateLimitBurst + } + + return max(int(math.Ceil(c.GetRateLimit()*defaultRateLimitBurstFactor)), 1) +} + +// GetRateLimitMaxBuckets returns the maximum number of wallet buckets the limiter may keep in +// memory. It returns 0 when not configured, which selects the limiter's own default. +func (c *Config) GetRateLimitMaxBuckets() int { + return max(c.RateLimitMaxBuckets, 0) +} diff --git a/token/services/selector/config/driver_test.go b/token/services/selector/config/driver_test.go index a25d771d83..bc2b1a032c 100644 --- a/token/services/selector/config/driver_test.go +++ b/token/services/selector/config/driver_test.go @@ -13,6 +13,7 @@ import ( "github.com/LFDT-Panurus/panurus/token/services/selector/config/mock" "github.com/LFDT-Panurus/panurus/token/services/selector/driver" + "github.com/LFDT-Panurus/panurus/token/services/selector/ratelimit" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -228,6 +229,104 @@ func TestConfig_GetFetcherCacheMaxQueries(t *testing.T) { } } +// TestConfig_RateLimit verifies the built-in selection rate limiter is off unless it is enabled +// explicitly or implied by a positive rate, and that the documented defaults apply once it is on. +func TestConfig_RateLimit(t *testing.T) { + tests := []struct { + name string + config *Config + expectedEnabled bool + expectedRate float64 + expectedBurst int + }{ + { + name: "disabled by default", + config: &Config{}, + expectedEnabled: false, + expectedRate: 0, + expectedBurst: 0, + }, + { + name: "enabled without values uses defaults", + config: &Config{RateLimitEnabled: true}, + expectedEnabled: true, + expectedRate: defaultRateLimit, + expectedBurst: defaultRateLimit * defaultRateLimitBurstFactor, + }, + { + name: "positive rate implies enabled", + config: &Config{RateLimit: 20}, + expectedEnabled: true, + expectedRate: 20, + expectedBurst: 40, + }, + { + name: "explicit burst is honoured", + config: &Config{RateLimit: 20, RateLimitBurst: 5}, + expectedEnabled: true, + expectedRate: 20, + expectedBurst: 5, + }, + { + name: "burst without rate defaults the rate", + config: &Config{RateLimitEnabled: true, RateLimitBurst: 5}, + expectedEnabled: true, + expectedRate: defaultRateLimit, + expectedBurst: 5, + }, + { + name: "fractional rate rounds the burst up", + config: &Config{RateLimit: 0.5}, + expectedEnabled: true, + expectedRate: 0.5, + expectedBurst: 1, + }, + { + name: "non-positive rate alone does not enable", + config: &Config{RateLimit: -1}, + expectedEnabled: false, + expectedRate: 0, + expectedBurst: 0, + }, + { + name: "non-positive rate with explicit enable falls back to the default", + config: &Config{RateLimitEnabled: true, RateLimit: -1}, + expectedEnabled: true, + expectedRate: defaultRateLimit, + expectedBurst: defaultRateLimit * defaultRateLimitBurstFactor, + }, + { + name: "non-positive burst falls back to the default", + config: &Config{RateLimit: 10, RateLimitBurst: -5}, + expectedEnabled: true, + expectedRate: 10, + expectedBurst: 20, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expectedEnabled, tt.config.IsRateLimitEnabled()) + assert.InDelta(t, tt.expectedRate, tt.config.GetRateLimit(), 0) + assert.Equal(t, tt.expectedBurst, tt.config.GetRateLimitBurst()) + }) + } +} + +// TestConfig_GetRateLimitMaxBuckets verifies the bucket cap defaults to zero, which lets the +// limiter pick its own, and that a negative value is treated the same way. +func TestConfig_GetRateLimitMaxBuckets(t *testing.T) { + assert.Equal(t, 0, (&Config{}).GetRateLimitMaxBuckets()) + assert.Equal(t, 0, (&Config{RateLimitMaxBuckets: -1}).GetRateLimitMaxBuckets()) + assert.Equal(t, 1024, (&Config{RateLimitMaxBuckets: 1024}).GetRateLimitMaxBuckets()) +} + +// TestConfig_ImplementsRateLimitConfiguration makes sure the parsed configuration keeps satisfying +// the interface the limiter builds itself from. +func TestConfig_ImplementsRateLimitConfiguration(t *testing.T) { + var _ ratelimit.Configuration = &Config{} +} + // TestNew verifies config parsing handles valid configs, empty configs, and unmarshal errors. func TestNew(t *testing.T) { tests := []struct { diff --git a/token/services/selector/ratelimit/decorator.go b/token/services/selector/ratelimit/decorator.go new file mode 100644 index 0000000000..e3a7091a13 --- /dev/null +++ b/token/services/selector/ratelimit/decorator.go @@ -0,0 +1,95 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package ratelimit + +import ( + "context" + + "github.com/LFDT-Panurus/panurus/token" + token2 "github.com/LFDT-Panurus/panurus/token/token" +) + +// Decorate returns a token.SelectorManager that meters every selection request performed by the +// selectors delegate hands out. scope identifies the token management service the manager belongs +// to (its TMS id), so wallets of different networks or namespaces get separate allowances. +// +// Only Selector.Select is metered. Unlock and Close pass straight through: releasing tokens must +// never be throttled, or a throttled wallet could not clean up after itself. +// +// When limiter is nil, delegate is returned unchanged. Rate limiting is disabled by default, and +// this keeps the disabled path free of any wrapper. +// +// The returned manager does not own limiter: closing or stopping the manager leaves the limiter +// and the allowances it tracks untouched, which is what makes it safe to share one limiter across +// the managers a selector service recreates on every public-parameter reload. +func Decorate(delegate token.SelectorManager, limiter Limiter, scope string) token.SelectorManager { + if limiter == nil { + return delegate + } + + return &manager{delegate: delegate, limiter: limiter, scope: scope} +} + +// manager decorates a token.SelectorManager with per-wallet metering of selection requests. +type manager struct { + delegate token.SelectorManager + limiter Limiter + scope string +} + +// NewSelector returns a selector bound to the passed transaction id whose Select calls are +// metered by the manager's limiter. +func (m *manager) NewSelector(id string) (token.Selector, error) { + delegate, err := m.delegate.NewSelector(id) + if err != nil { + return nil, err + } + + return &selector{delegate: delegate, limiter: m.limiter, scope: m.scope}, nil +} + +// Unlock unlocks the tokens bound to the passed id. It is never metered. +func (m *manager) Unlock(ctx context.Context, id string) error { + return m.delegate.Unlock(ctx, id) +} + +// Close closes the selector bound to the passed id and releases its resources. It is never +// metered, and it does not stop the limiter. +func (m *manager) Close(id string) error { + return m.delegate.Close(id) +} + +// selector decorates a token.Selector, charging one request to the wallet's bucket per Select +// call. The whole call counts as one request no matter how many tokens it locks or how many times +// it retries internally. +type selector struct { + delegate token.Selector + limiter Limiter + scope string +} + +// Select meters the request against the owner's allowance and, if allowed, delegates the actual +// selection. When the allowance is exhausted it returns the limiter's error, which wraps +// token.SelectorRateLimited, without touching the underlying selector: nothing is locked, so +// there is nothing to release. +func (s *selector) Select(ctx context.Context, ownerFilter token.OwnerFilter, q string, tokenType token2.Type) ([]*token2.ID, token2.Quantity, error) { + var walletID string + if ownerFilter != nil { + walletID = ownerFilter.ID() + } + if err := s.limiter.Allow(ctx, s.scope, walletID); err != nil { + return nil, nil, err + } + + return s.delegate.Select(ctx, ownerFilter, q, tokenType) +} + +// Close closes the underlying selector. It does not stop the limiter, which outlives the +// selectors it meters. +func (s *selector) Close() error { + return s.delegate.Close() +} diff --git a/token/services/selector/ratelimit/decorator_test.go b/token/services/selector/ratelimit/decorator_test.go new file mode 100644 index 0000000000..ef84d15581 --- /dev/null +++ b/token/services/selector/ratelimit/decorator_test.go @@ -0,0 +1,250 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package ratelimit + +import ( + "context" + "sync" + "testing" + + "github.com/LFDT-Panurus/panurus/token" + token2 "github.com/LFDT-Panurus/panurus/token/token" + "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// wallet is a token.OwnerFilter identified by a wallet id. +type wallet string + +func (w wallet) ID() string { return string(w) } + +// fakeSelector records what it was asked to do. +type fakeSelector struct { + mu sync.Mutex + selectCalls int + closeCalls int + err error +} + +func (s *fakeSelector) Select(_ context.Context, _ token.OwnerFilter, _ string, _ token2.Type) ([]*token2.ID, token2.Quantity, error) { + s.mu.Lock() + defer s.mu.Unlock() + + s.selectCalls++ + if s.err != nil { + return nil, nil, s.err + } + + return []*token2.ID{{TxId: "tx", Index: 0}}, token2.NewZeroQuantity(64), nil +} + +func (s *fakeSelector) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + + s.closeCalls++ + + return nil +} + +func (s *fakeSelector) calls() (int, int) { + s.mu.Lock() + defer s.mu.Unlock() + + return s.selectCalls, s.closeCalls +} + +// fakeManager hands out one fakeSelector and records the unlock and close calls it receives. +type fakeManager struct { + selector *fakeSelector + newSelectErr error + unlockCalls int + closeCalls int +} + +func newFakeManager() *fakeManager { + return &fakeManager{selector: &fakeSelector{}} +} + +func (m *fakeManager) NewSelector(string) (token.Selector, error) { + if m.newSelectErr != nil { + return nil, m.newSelectErr + } + + return m.selector, nil +} + +func (m *fakeManager) Unlock(context.Context, string) error { + m.unlockCalls++ + + return nil +} + +func (m *fakeManager) Close(string) error { + m.closeCalls++ + + return nil +} + +// countingLimiter counts every metering decision, so tests can tell what was metered and what was +// not. +type countingLimiter struct { + calls int + wallets []string + err error +} + +func (l *countingLimiter) Allow(_ context.Context, _ string, walletID string) error { + l.calls++ + l.wallets = append(l.wallets, walletID) + + return l.err +} + +// TestDecorate_NilLimiterIsPassthrough verifies the default, disabled configuration adds no +// wrapper at all. +func TestDecorate_NilLimiterIsPassthrough(t *testing.T) { + delegate := newFakeManager() + + assert.Same(t, delegate, Decorate(delegate, nil, testScope)) +} + +// TestDecorate_AllowedSelectionReachesDelegate verifies an allowed request is forwarded untouched, +// and that exactly one request is charged per Select call. +func TestDecorate_AllowedSelectionReachesDelegate(t *testing.T) { + delegate := newFakeManager() + limiter := &countingLimiter{} + mgr := Decorate(delegate, limiter, testScope) + + selector, err := mgr.NewSelector("tx1") + require.NoError(t, err) + + // Building the selector must not consume any allowance. + assert.Equal(t, 0, limiter.calls) + + ids, sum, err := selector.Select(context.Background(), wallet("alice"), "10", "USD") + require.NoError(t, err) + assert.Len(t, ids, 1) + assert.NotNil(t, sum) + + selectCalls, _ := delegate.selector.calls() + assert.Equal(t, 1, selectCalls) + assert.Equal(t, 1, limiter.calls) + assert.Equal(t, []string{"alice"}, limiter.wallets) +} + +// TestDecorate_DeniedSelectionNeverReachesDelegate verifies a denied request fails fast with the +// token.SelectorRateLimited contract error and does not touch the underlying selector, so no token +// is locked and there is nothing to leak. +func TestDecorate_DeniedSelectionNeverReachesDelegate(t *testing.T) { + delegate := newFakeManager() + limiter := &countingLimiter{err: errors.Wrapf(token.SelectorRateLimited, "wallet [alice] throttled")} + mgr := Decorate(delegate, limiter, testScope) + + selector, err := mgr.NewSelector("tx1") + require.NoError(t, err) + + ids, sum, err := selector.Select(context.Background(), wallet("alice"), "10", "USD") + require.ErrorIs(t, err, token.SelectorRateLimited) + assert.Nil(t, ids) + assert.Nil(t, sum) + + selectCalls, _ := delegate.selector.calls() + assert.Equal(t, 0, selectCalls, "the delegate must not run, so nothing gets locked") +} + +// TestDecorate_UnlockAndCloseAreNotMetered verifies releasing tokens is never throttled: a wallet +// that has exhausted its allowance must still be able to clean up. +func TestDecorate_UnlockAndCloseAreNotMetered(t *testing.T) { + delegate := newFakeManager() + limiter := &countingLimiter{err: errors.Wrapf(token.SelectorRateLimited, "throttled")} + mgr := Decorate(delegate, limiter, testScope) + + selector, err := mgr.NewSelector("tx1") + require.NoError(t, err) + + require.NoError(t, mgr.Unlock(context.Background(), "tx1")) + require.NoError(t, mgr.Close("tx1")) + require.NoError(t, selector.Close()) + + assert.Equal(t, 0, limiter.calls) + assert.Equal(t, 1, delegate.unlockCalls) + assert.Equal(t, 1, delegate.closeCalls) + _, closeCalls := delegate.selector.calls() + assert.Equal(t, 1, closeCalls) +} + +// TestDecorate_NilOwnerFilterIsNotThrottled verifies a request without an owner filter is metered +// as an empty wallet, which the built-in limiter lets through, and reaches the delegate that +// rejects it on its own terms. +func TestDecorate_NilOwnerFilterIsNotThrottled(t *testing.T) { + delegate := newFakeManager() + delegate.selector.err = errors.New("no owner filter specified") + mgr := Decorate(delegate, New(Config{Rate: 1, Burst: 1}), testScope) + + selector, err := mgr.NewSelector("tx1") + require.NoError(t, err) + + for range 10 { + _, _, err = selector.Select(context.Background(), nil, "10", "USD") + require.ErrorContains(t, err, "no owner filter specified") + } + + selectCalls, _ := delegate.selector.calls() + assert.Equal(t, 10, selectCalls) +} + +// TestDecorate_NewSelectorError verifies a failure to build the underlying selector is propagated. +func TestDecorate_NewSelectorError(t *testing.T) { + delegate := newFakeManager() + delegate.newSelectErr = errors.New("no selector for you") + mgr := Decorate(delegate, &countingLimiter{}, testScope) + + selector, err := mgr.NewSelector("tx1") + require.ErrorContains(t, err, "no selector for you") + assert.Nil(t, selector) +} + +// TestDecorate_MetersPerSelectCallNotPerLock verifies the whole selection counts as one request: +// a Select call that locks many tokens, or retries internally, is charged exactly once. +func TestDecorate_MetersPerSelectCallNotPerLock(t *testing.T) { + delegate := newFakeManager() + limiter := New(Config{Rate: 0.001, Burst: 3}) + mgr := Decorate(delegate, limiter, testScope) + + selector, err := mgr.NewSelector("tx1") + require.NoError(t, err) + + // The fake selector stands in for a selection that locks several tokens and retries: three + // Select calls fit in a burst of three, whatever happens inside them. + for range 3 { + _, _, err = selector.Select(context.Background(), wallet("alice"), "10", "USD") + require.NoError(t, err) + } + _, _, err = selector.Select(context.Background(), wallet("alice"), "10", "USD") + require.ErrorIs(t, err, token.SelectorRateLimited) +} + +// TestDecorate_WalletsAreIndependent verifies two wallets selecting through the same manager do not +// share an allowance. +func TestDecorate_WalletsAreIndependent(t *testing.T) { + delegate := newFakeManager() + mgr := Decorate(delegate, New(Config{Rate: 0.001, Burst: 1}), testScope) + + selector, err := mgr.NewSelector("tx1") + require.NoError(t, err) + ctx := context.Background() + + _, _, err = selector.Select(ctx, wallet("alice"), "10", "USD") + require.NoError(t, err) + _, _, err = selector.Select(ctx, wallet("alice"), "10", "USD") + require.ErrorIs(t, err, token.SelectorRateLimited) + + _, _, err = selector.Select(ctx, wallet("bob"), "10", "USD") + require.NoError(t, err) +} diff --git a/token/services/selector/ratelimit/limiter.go b/token/services/selector/ratelimit/limiter.go new file mode 100644 index 0000000000..df96e1c0aa --- /dev/null +++ b/token/services/selector/ratelimit/limiter.go @@ -0,0 +1,328 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +// Package ratelimit provides an opt-in, built-in per-wallet rate limiter for token selection. +// +// The limiter is disabled by default. It is activated either from configuration +// (the token.selector.rateLimit* keys, see token/services/selector/config) or programmatically +// through the functional options in this package (WithLimiter, WithDefaultLimiter). +// +// Metering happens once per selection request, that is once per Selector.Select call, not once +// per token lock attempt. Decorate wires this up by wrapping a token.SelectorManager. Charging +// per lock attempt would let the internal contention retries of the selectors drain a wallet's +// allowance, and would charge a large selection more than a small one. +// +// When a wallet exceeds its allowance, the limiter returns an error wrapping +// token.SelectorRateLimited, the fail-fast contract both selector drivers already honour: the +// selection aborts immediately, no tokens stay locked, and the error reaches the caller. +package ratelimit + +import ( + "context" + "math" + "slices" + "sync" + "time" + + "github.com/LFDT-Panurus/panurus/token" + "github.com/LFDT-Panurus/panurus/token/services/logging" + "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" +) + +const ( + // DefaultRate is the per-wallet selection rate, in requests per second, used when rate + // limiting is enabled without an explicit rate. It is high enough not to interfere with + // normal interactive workloads while still capping a runaway or abusive client. + DefaultRate = 100.0 + // DefaultBurstFactor multiplies the rate to obtain the default burst capacity, so the + // default configuration allows DefaultRate*DefaultBurstFactor back-to-back requests. + DefaultBurstFactor = 2 + // DefaultIdleTimeout is how long a wallet bucket may stay untouched before it becomes + // eligible for eviction. + DefaultIdleTimeout = 5 * time.Minute + // DefaultMaxBuckets caps the number of live wallet buckets, bounding the memory a + // long-running node spends on transient wallet ids. + DefaultMaxBuckets = 4096 + + // maxIdleTimeout caps the idle timeout derived from a very slow rate (see normalize). + maxIdleTimeout = time.Hour + // sweepEveryNCalls is the number of Allow calls between two amortized sweeps of idle + // buckets. Sweeping is O(number of buckets), so it must not run on every call. + sweepEveryNCalls = 512 + // evictionWarnInterval is the minimum time between two warnings about a limiter that sits + // at its bucket cap. At the cap every new wallet evicts one, so warning every time would + // flood the log. + evictionWarnInterval = time.Minute + // keySeparator joins the scope and the wallet id. It cannot appear in either, so two + // distinct (scope, wallet) pairs can never map to the same bucket. + keySeparator = "\x00" +) + +var logger = logging.MustGetLogger() + +// Limiter meters token selection requests per wallet. +// +// Implementations must be safe for concurrent use. +type Limiter interface { + // Allow reports whether a selection request for walletID within scope may proceed. + // scope isolates wallets belonging to different token management services, so that the + // same wallet id used in two networks or namespaces does not share one allowance. + // It returns nil when the request is allowed, and an error wrapping + // token.SelectorRateLimited when the request must be denied. + // An empty walletID is never throttled. + Allow(ctx context.Context, scope string, walletID string) error +} + +// Config configures the built-in token-bucket limiter. Non-positive fields fall back to the +// package defaults, so the zero Config is the default configuration. +type Config struct { + // Rate is the sustained number of selection requests allowed per second, per wallet. + Rate float64 + // Burst is the maximum number of selection requests a single wallet may issue + // back-to-back before it is throttled down to Rate. + Burst int + // IdleTimeout is how long a bucket may stay untouched before it becomes eligible for + // eviction. It is raised to at least the time a bucket needs to refill from empty. + IdleTimeout time.Duration + // MaxBuckets is the maximum number of live buckets. Once the idle sweep cannot bring the + // limiter back under this number, the least recently used buckets are dropped. + MaxBuckets int +} + +// DefaultConfig returns the configuration used when rate limiting is enabled without any +// explicit rate or burst: DefaultRate requests per second per wallet, with a burst of +// DefaultRate*DefaultBurstFactor. +func DefaultConfig() Config { + return Config{ + Rate: DefaultRate, + Burst: DefaultRate * DefaultBurstFactor, + IdleTimeout: DefaultIdleTimeout, + MaxBuckets: DefaultMaxBuckets, + } +} + +// normalize replaces non-positive fields with their defaults and raises IdleTimeout to at least +// the time a bucket needs to refill from empty to full. Evicting an idle bucket hands back a +// full bucket, so a shorter idle timeout would let a wallet skip part of the wait it owes simply +// by pausing. +func (c Config) normalize() Config { + if c.Rate <= 0 { + c.Rate = DefaultRate + } + if c.Burst <= 0 { + c.Burst = max(int(math.Ceil(c.Rate*DefaultBurstFactor)), 1) + } + if c.MaxBuckets <= 0 { + c.MaxBuckets = DefaultMaxBuckets + } + if c.IdleTimeout <= 0 { + c.IdleTimeout = DefaultIdleTimeout + } + if refill := refillDuration(float64(c.Burst), c.Rate); c.IdleTimeout < refill { + c.IdleTimeout = refill + } + + return c +} + +// refillDuration returns how long it takes to accumulate the given number of requests at rate +// requests per second, capped at maxIdleTimeout to keep the result a sane duration. +func refillDuration(requests float64, rate float64) time.Duration { + seconds := requests / rate + if seconds >= maxIdleTimeout.Seconds() { + return maxIdleTimeout + } + + return time.Duration(seconds * float64(time.Second)) +} + +// bucket is the per-wallet token bucket. It is only ever accessed with BucketLimiter.mu held. +type bucket struct { + // tokens is the number of selection requests currently available to the wallet. + tokens float64 + // last is the time tokens was last recomputed, and doubles as the wallet's last-seen + // time for eviction purposes. + last time.Time +} + +// refill adds the requests accrued since the last access, capping at burst. +func (b *bucket) refill(now time.Time, rate float64, burst float64) { + elapsed := now.Sub(b.last) + if elapsed > 0 { + b.tokens = math.Min(burst, b.tokens+elapsed.Seconds()*rate) + } + b.last = now +} + +// BucketLimiter is a thread-safe token-bucket Limiter keyed by (scope, wallet id). +// +// Buckets are allocated lazily on first use and pruned again without a background goroutine: +// every sweepEveryNCalls requests, and whenever the bucket count exceeds MaxBuckets, idle +// buckets are removed. This keeps the memory of a long-running node bounded even when it sees +// a large number of short-lived wallet ids. +type BucketLimiter struct { + rate float64 + burst float64 + idleTimeout time.Duration + maxBuckets int + // now returns the current time. It is a field so tests can drive the bucket with a + // deterministic clock instead of sleeping. + now func() time.Time + + mu sync.Mutex + buckets map[string]*bucket + callsSinceSweep int + lastEvictionWarn time.Time +} + +// New returns a BucketLimiter with the given configuration. Non-positive configuration values +// fall back to the package defaults, so New(Config{}) is the default limiter. +func New(c Config) *BucketLimiter { + c = c.normalize() + + return &BucketLimiter{ + rate: c.Rate, + burst: float64(c.Burst), + idleTimeout: c.IdleTimeout, + maxBuckets: c.MaxBuckets, + now: time.Now, + buckets: make(map[string]*bucket), + } +} + +// Allow consumes one request from the bucket of walletID within scope. It returns an error +// wrapping token.SelectorRateLimited when the wallet has no request left, and nil otherwise. +// An empty walletID is never throttled. +func (l *BucketLimiter) Allow(_ context.Context, scope string, walletID string) error { + if len(walletID) == 0 { + return nil + } + + now := l.now() + + l.mu.Lock() + defer l.mu.Unlock() + + l.maybeSweep(now) + + key := bucketKey(scope, walletID) + b, ok := l.buckets[key] + if !ok { + l.makeRoom(now) + // A new wallet starts with a full bucket, which is also what an evicted wallet gets + // back: see Config.normalize for why that is safe. + b = &bucket{tokens: l.burst, last: now} + l.buckets[key] = b + } else { + b.refill(now, l.rate, l.burst) + } + + if b.tokens < 1 { + retryAfter := refillDuration(1-b.tokens, l.rate) + + return errors.Wrapf( + token.SelectorRateLimited, + "wallet [%s] exceeded the selection rate of %g requests/s (burst %g) for tms [%s], retry in %s", + walletID, l.rate, l.burst, scope, retryAfter, + ) + } + b.tokens-- + + return nil +} + +// Stop releases the memory held by the limiter. The limiter stays usable afterwards, with every +// wallet starting from a full bucket again. +// +// A selector service never calls this: its Shutdown also runs on routine public-parameter +// reloads, and resetting every wallet's allowance there would let a throttled client wash out +// its debt. Stop is for callers that own a limiter and are done with it. +func (l *BucketLimiter) Stop() { + l.mu.Lock() + defer l.mu.Unlock() + + l.buckets = make(map[string]*bucket) + l.callsSinceSweep = 0 +} + +// BucketCount returns the number of live buckets. It is exported for tests and diagnostics. +func (l *BucketLimiter) BucketCount() int { + l.mu.Lock() + defer l.mu.Unlock() + + return len(l.buckets) +} + +// maybeSweep prunes idle buckets once every sweepEveryNCalls calls. Sweeping is linear in the +// number of buckets, so spreading it over many calls keeps Allow constant-time on average. +// It must be called with l.mu held. +func (l *BucketLimiter) maybeSweep(now time.Time) { + l.callsSinceSweep++ + if l.callsSinceSweep < sweepEveryNCalls { + return + } + l.callsSinceSweep = 0 + l.sweepIdle(now) +} + +// makeRoom guarantees there is room for one more bucket, so the limiter never holds more than +// maxBuckets of them. It first tries the cheap route of dropping idle buckets, and only evicts +// live ones if that was not enough. It must be called with l.mu held. +func (l *BucketLimiter) makeRoom(now time.Time) { + if len(l.buckets) < l.maxBuckets { + return + } + + l.sweepIdle(now) + if excess := len(l.buckets) - l.maxBuckets + 1; excess > 0 { + l.evictLeastRecentlyUsed(now, excess) + } +} + +// sweepIdle removes every bucket that has not been used for idleTimeout. It must be called with +// l.mu held. +func (l *BucketLimiter) sweepIdle(now time.Time) { + for key, b := range l.buckets { + if now.Sub(b.last) >= l.idleTimeout { + delete(l.buckets, key) + } + } +} + +// evictLeastRecentlyUsed drops the n buckets that were accessed longest ago. It runs only when +// pruning idle buckets was not enough to stay within maxBuckets, which means the node is tracking +// more active wallets than it is configured for: bounding memory takes precedence over remembering +// every wallet's debt. It must be called with l.mu held. +func (l *BucketLimiter) evictLeastRecentlyUsed(now time.Time, n int) { + // At the cap, every new wallet evicts one, so this is throttled to keep the log readable. + if now.Sub(l.lastEvictionWarn) >= evictionWarnInterval { + l.lastEvictionWarn = now + logger.Warnf( + "selection rate limiter is at its maximum of %d buckets and is evicting the least recently used "+ + "wallets, which resets their allowance: consider raising token.selector.rateLimitMaxBuckets", + l.maxBuckets, + ) + } else { + logger.Debugf("selection rate limiter evicting %d least recently used buckets of %d", n, len(l.buckets)) + } + + keys := make([]string, 0, len(l.buckets)) + for key := range l.buckets { + keys = append(keys, key) + } + slices.SortFunc(keys, func(a, b string) int { + return l.buckets[a].last.Compare(l.buckets[b].last) + }) + for _, key := range keys[:n] { + delete(l.buckets, key) + } +} + +// bucketKey scopes a wallet id to its token management service, so the same wallet id in two +// networks or namespaces does not share an allowance. +func bucketKey(scope string, walletID string) string { + return scope + keySeparator + walletID +} diff --git a/token/services/selector/ratelimit/limiter_test.go b/token/services/selector/ratelimit/limiter_test.go new file mode 100644 index 0000000000..309c36c553 --- /dev/null +++ b/token/services/selector/ratelimit/limiter_test.go @@ -0,0 +1,329 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package ratelimit + +import ( + "context" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/LFDT-Panurus/panurus/token" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testScope = "testnet,testchannel,testns" + +// fakeClock is a manually advanced clock, so the bucket arithmetic can be tested without sleeping. +type fakeClock struct { + mu sync.Mutex + now time.Time +} + +func newFakeClock() *fakeClock { + return &fakeClock{now: time.Date(2026, time.August, 18, 12, 0, 0, 0, time.UTC)} +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + + return c.now +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + + c.now = c.now.Add(d) +} + +// newTestLimiter returns a limiter driven by the returned fake clock. +func newTestLimiter(t *testing.T, c Config) (*BucketLimiter, *fakeClock) { + t.Helper() + + clock := newFakeClock() + l := New(c) + l.now = clock.Now + + return l, clock +} + +// TestBucketLimiter_BurstThenDeny verifies a wallet may issue exactly Burst requests back-to-back +// and is denied afterwards, with an error wrapping token.SelectorRateLimited. +func TestBucketLimiter_BurstThenDeny(t *testing.T) { + l, _ := newTestLimiter(t, Config{Rate: 10, Burst: 3}) + ctx := context.Background() + + for i := range 3 { + require.NoError(t, l.Allow(ctx, testScope, "alice"), "request %d must be allowed", i) + } + + err := l.Allow(ctx, testScope, "alice") + require.Error(t, err) + require.ErrorIs(t, err, token.SelectorRateLimited) + assert.Contains(t, err.Error(), "alice") + assert.Contains(t, err.Error(), testScope) +} + +// TestBucketLimiter_Refill verifies the bucket refills at the configured rate and never beyond +// the burst capacity. +func TestBucketLimiter_Refill(t *testing.T) { + // 10 requests/s means one request every 100ms. + l, clock := newTestLimiter(t, Config{Rate: 10, Burst: 2}) + ctx := context.Background() + + require.NoError(t, l.Allow(ctx, testScope, "alice")) + require.NoError(t, l.Allow(ctx, testScope, "alice")) + require.ErrorIs(t, l.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) + + // Not enough time for a whole request yet. + clock.Advance(50 * time.Millisecond) + require.ErrorIs(t, l.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) + + // A further 50ms completes the first refilled request. + clock.Advance(50 * time.Millisecond) + require.NoError(t, l.Allow(ctx, testScope, "alice")) + require.ErrorIs(t, l.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) + + // A long pause refills at most Burst requests, not more. + clock.Advance(time.Hour) + require.NoError(t, l.Allow(ctx, testScope, "alice")) + require.NoError(t, l.Allow(ctx, testScope, "alice")) + require.ErrorIs(t, l.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) +} + +// TestBucketLimiter_SustainedRate verifies that, once the burst is spent, a wallet is served at +// the configured sustained rate. +func TestBucketLimiter_SustainedRate(t *testing.T) { + l, clock := newTestLimiter(t, Config{Rate: 4, Burst: 1}) + ctx := context.Background() + + require.NoError(t, l.Allow(ctx, testScope, "alice")) + for range 10 { + require.ErrorIs(t, l.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) + clock.Advance(250 * time.Millisecond) // exactly one request at 4 requests/s + require.NoError(t, l.Allow(ctx, testScope, "alice")) + } +} + +// TestBucketLimiter_EmptyWalletBypass verifies an empty wallet id is never throttled and does not +// allocate a bucket. +func TestBucketLimiter_EmptyWalletBypass(t *testing.T) { + l, _ := newTestLimiter(t, Config{Rate: 1, Burst: 1}) + ctx := context.Background() + + for range 100 { + require.NoError(t, l.Allow(ctx, testScope, "")) + } + assert.Equal(t, 0, l.BucketCount()) +} + +// TestBucketLimiter_WalletIsolation verifies one wallet exhausting its allowance does not affect +// another. +func TestBucketLimiter_WalletIsolation(t *testing.T) { + l, _ := newTestLimiter(t, Config{Rate: 1, Burst: 1}) + ctx := context.Background() + + require.NoError(t, l.Allow(ctx, testScope, "alice")) + require.ErrorIs(t, l.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) + + require.NoError(t, l.Allow(ctx, testScope, "bob")) + require.ErrorIs(t, l.Allow(ctx, testScope, "bob"), token.SelectorRateLimited) +} + +// TestBucketLimiter_ScopeIsolation verifies the same wallet id in two token management services +// gets two independent allowances, and that scope and wallet id cannot be confused for one +// another. +func TestBucketLimiter_ScopeIsolation(t *testing.T) { + l, _ := newTestLimiter(t, Config{Rate: 1, Burst: 1}) + ctx := context.Background() + + require.NoError(t, l.Allow(ctx, "net1,ch1,ns1", "alice")) + require.ErrorIs(t, l.Allow(ctx, "net1,ch1,ns1", "alice"), token.SelectorRateLimited) + + require.NoError(t, l.Allow(ctx, "net2,ch1,ns1", "alice")) + require.Equal(t, 2, l.BucketCount()) + + // ("a", "bc") and ("ab", "c") must not collide. + require.NoError(t, l.Allow(ctx, "a", "bc")) + require.NoError(t, l.Allow(ctx, "ab", "c")) + require.Equal(t, 4, l.BucketCount()) +} + +// TestBucketLimiter_ConcurrentWallets verifies that under concurrency each wallet gets exactly its +// own allowance: no cross-wallet interference, and no double spending of a bucket. +func TestBucketLimiter_ConcurrentWallets(t *testing.T) { + const ( + wallets = 16 + goroutinesPerWalet = 8 + requestsPerRoutine = 25 + burst = 40 + ) + + // Rate 0 would be replaced by the default, so use a rate slow enough that the fake clock, + // which never advances, cannot refill anything. + l, _ := newTestLimiter(t, Config{Rate: 0.0001, Burst: burst}) + ctx := context.Background() + + allowed := make([]atomic.Int64, wallets) + var wg sync.WaitGroup + for w := range wallets { + walletID := "wallet-" + strconv.Itoa(w) + for range goroutinesPerWalet { + wg.Go(func() { + for range requestsPerRoutine { + if err := l.Allow(ctx, testScope, walletID); err == nil { + allowed[w].Add(1) + } else { + assert.ErrorIs(t, err, token.SelectorRateLimited) + } + } + }) + } + } + wg.Wait() + + // Every wallet asks for more than its burst, so each must be served exactly burst times. + require.Greater(t, goroutinesPerWalet*requestsPerRoutine, burst, "the test must ask for more than the burst") + for w := range wallets { + assert.Equal(t, int64(burst), allowed[w].Load(), "wallet %d", w) + } + assert.Equal(t, wallets, l.BucketCount()) +} + +// TestBucketLimiter_IdleEviction verifies idle buckets are pruned during ordinary access, so a +// node that sees many short-lived wallet ids does not accumulate them. +func TestBucketLimiter_IdleEviction(t *testing.T) { + l, clock := newTestLimiter(t, Config{Rate: 100, Burst: 100, IdleTimeout: time.Minute}) + ctx := context.Background() + + // A batch of one-shot wallets, none of which is idle yet, so all of them are kept. + for i := range sweepEveryNCalls { + require.NoError(t, l.Allow(ctx, testScope, "transient-"+strconv.Itoa(i))) + } + assert.Equal(t, sweepEveryNCalls, l.BucketCount()) + + // They all go idle, and the sweep that follows within the next sweepEveryNCalls requests + // of an active wallet removes them, leaving only that wallet's bucket behind. + clock.Advance(2 * time.Minute) + for range sweepEveryNCalls { + // The active wallet is throttled along the way, which is beside the point here. + _ = l.Allow(ctx, testScope, "alice") + } + assert.Equal(t, 1, l.BucketCount()) +} + +// TestBucketLimiter_BoundedUnderWalletTurnover verifies the bucket count stays bounded when a +// long-running node keeps seeing new wallet ids without any of them going idle. +func TestBucketLimiter_BoundedUnderWalletTurnover(t *testing.T) { + const maxBuckets = 64 + + l, clock := newTestLimiter(t, Config{Rate: 100, Burst: 100, IdleTimeout: time.Hour, MaxBuckets: maxBuckets}) + ctx := context.Background() + + for i := range 20 * maxBuckets { + require.NoError(t, l.Allow(ctx, testScope, "wallet-"+strconv.Itoa(i))) + // No bucket ever reaches the idle timeout, so only the hard cap bounds memory. + clock.Advance(time.Millisecond) + require.LessOrEqual(t, l.BucketCount(), maxBuckets, "bucket count must stay bounded") + } +} + +// TestBucketLimiter_EvictsLeastRecentlyUsed verifies that, when the idle sweep is not enough, it +// is the least recently used buckets that go. +func TestBucketLimiter_EvictsLeastRecentlyUsed(t *testing.T) { + // A rate slow enough that the seconds the clock advances refill nothing measurable. + l, clock := newTestLimiter(t, Config{Rate: 0.001, Burst: 1, IdleTimeout: time.Hour, MaxBuckets: 2}) + ctx := context.Background() + + require.NoError(t, l.Allow(ctx, testScope, "old")) + clock.Advance(time.Second) + require.NoError(t, l.Allow(ctx, testScope, "recent")) + clock.Advance(time.Second) + + // "recent" is out of allowance and stays out: it was not the one evicted. + require.ErrorIs(t, l.Allow(ctx, testScope, "recent"), token.SelectorRateLimited) + clock.Advance(time.Second) + + // A third wallet would take the limiter past its cap of 2, so the oldest bucket goes. + require.NoError(t, l.Allow(ctx, testScope, "new")) + assert.Equal(t, 2, l.BucketCount()) + require.ErrorIs(t, l.Allow(ctx, testScope, "recent"), token.SelectorRateLimited) + // "old" was evicted, so it starts over with a full bucket. + require.NoError(t, l.Allow(ctx, testScope, "old")) +} + +// TestBucketLimiter_IdleTimeoutFloor verifies the configured idle timeout is raised to at least +// the time a bucket needs to refill, so eviction cannot be used to skip the wait. +func TestBucketLimiter_IdleTimeoutFloor(t *testing.T) { + // Refilling 10 requests at 1 request/s takes 10s, far more than the configured 1ms. + c := Config{Rate: 1, Burst: 10, IdleTimeout: time.Millisecond}.normalize() + assert.Equal(t, 10*time.Second, c.IdleTimeout) + + // An absurdly slow rate is capped instead of producing a nonsensical duration. + c = Config{Rate: 1e-9, Burst: 1}.normalize() + assert.Equal(t, maxIdleTimeout, c.IdleTimeout) +} + +// TestBucketLimiter_Stop verifies Stop releases the buckets and leaves the limiter usable. +func TestBucketLimiter_Stop(t *testing.T) { + l, _ := newTestLimiter(t, Config{Rate: 1, Burst: 1}) + ctx := context.Background() + + require.NoError(t, l.Allow(ctx, testScope, "alice")) + require.ErrorIs(t, l.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) + assert.Equal(t, 1, l.BucketCount()) + + l.Stop() + assert.Equal(t, 0, l.BucketCount()) + require.NoError(t, l.Allow(ctx, testScope, "alice")) +} + +// TestDefaultConfig verifies the defaults documented for the enabled-without-values case. +func TestDefaultConfig(t *testing.T) { + c := DefaultConfig() + assert.InDelta(t, 100.0, c.Rate, 0) + assert.Equal(t, 200, c.Burst) + assert.Equal(t, DefaultIdleTimeout, c.IdleTimeout) + assert.Equal(t, DefaultMaxBuckets, c.MaxBuckets) +} + +// TestConfig_Normalize verifies non-positive configuration values fall back to the defaults, so +// that the zero Config is the default configuration. +func TestConfig_Normalize(t *testing.T) { + c := Config{}.normalize() + assert.InDelta(t, DefaultRate, c.Rate, 0) + assert.Equal(t, 200, c.Burst) + assert.Equal(t, DefaultMaxBuckets, c.MaxBuckets) + assert.Equal(t, DefaultIdleTimeout, c.IdleTimeout) + + c = Config{Rate: -1, Burst: -1, MaxBuckets: -1, IdleTimeout: -time.Second}.normalize() + assert.InDelta(t, DefaultRate, c.Rate, 0) + assert.Equal(t, 200, c.Burst) + assert.Equal(t, DefaultMaxBuckets, c.MaxBuckets) + assert.Equal(t, DefaultIdleTimeout, c.IdleTimeout) + + // A fractional rate still yields a burst of at least one request. + c = Config{Rate: 0.1}.normalize() + assert.Equal(t, 1, c.Burst) +} + +// TestBucketLimiter_RetryAfterInError verifies the denial error tells the caller how long to wait. +func TestBucketLimiter_RetryAfterInError(t *testing.T) { + l, _ := newTestLimiter(t, Config{Rate: 2, Burst: 1}) + ctx := context.Background() + + require.NoError(t, l.Allow(ctx, testScope, "alice")) + err := l.Allow(ctx, testScope, "alice") + require.ErrorIs(t, err, token.SelectorRateLimited) + // At 2 requests/s, one request is worth 500ms. + assert.Contains(t, err.Error(), "retry in 500ms") +} diff --git a/token/services/selector/ratelimit/options.go b/token/services/selector/ratelimit/options.go new file mode 100644 index 0000000000..deb22c62e2 --- /dev/null +++ b/token/services/selector/ratelimit/options.go @@ -0,0 +1,86 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package ratelimit + +// Configuration is the subset of the selector configuration that describes the built-in rate +// limiter. It is implemented by token/services/selector/config.Config. +type Configuration interface { + // IsRateLimitEnabled tells whether the built-in per-wallet selection rate limiter is on. + IsRateLimitEnabled() bool + // GetRateLimit returns the allowed selection requests per second per wallet. + GetRateLimit() float64 + // GetRateLimitBurst returns the burst capacity of a wallet bucket. + GetRateLimitBurst() int + // GetRateLimitMaxBuckets returns the maximum number of live wallet buckets. + GetRateLimitMaxBuckets() int +} + +// Option customizes how a selector service obtains its rate limiter. Options take precedence +// over configuration. +type Option func(*Options) + +// Options collects the effect of the Option values passed to a selector service. +type Options struct { + limiter Limiter + // set records that an Option explicitly decided the limiter, including the decision to + // have none (WithLimiter(nil)). Without it, an explicit "off" would be + // indistinguishable from "not specified" and configuration would win. + set bool +} + +// WithLimiter makes the selector service use the passed limiter, ignoring the +// token.selector.rateLimit* configuration. Pass nil to disable rate limiting outright, whatever +// the configuration says. +// +// The service does not take ownership of the limiter: it never stops it, not even from Shutdown. +// A limiter passed here may therefore be shared between services, and its lifecycle stays with +// the caller. +func WithLimiter(limiter Limiter) Option { + return func(o *Options) { + o.limiter = limiter + o.set = true + } +} + +// WithDefaultLimiter enables the built-in per-wallet limiter with DefaultConfig, whatever the +// token.selector.rateLimit* configuration says. +func WithDefaultLimiter() Option { + return WithLimiter(New(DefaultConfig())) +} + +// CompileOptions applies the passed options. +func CompileOptions(opts ...Option) *Options { + o := &Options{} + for _, opt := range opts { + if opt != nil { + opt(o) + } + } + + return o +} + +// Limiter returns the limiter a selector service must use given its options and configuration, +// or nil when selection must not be rate limited. Options win over configuration; without any +// option, the limiter comes from the configuration, which has rate limiting off by default. +// +// A nil Configuration is treated as "not configured", so that a service whose configuration +// failed to parse still starts, with rate limiting off. +func (o *Options) Limiter(c Configuration) Limiter { + if o.set { + return o.limiter + } + if c == nil || !c.IsRateLimitEnabled() { + return nil + } + + return New(Config{ + Rate: c.GetRateLimit(), + Burst: c.GetRateLimitBurst(), + MaxBuckets: c.GetRateLimitMaxBuckets(), + }) +} diff --git a/token/services/selector/ratelimit/options_test.go b/token/services/selector/ratelimit/options_test.go new file mode 100644 index 0000000000..75dfa395e6 --- /dev/null +++ b/token/services/selector/ratelimit/options_test.go @@ -0,0 +1,90 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package ratelimit + +import ( + "context" + "testing" + + "github.com/LFDT-Panurus/panurus/token" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeConfiguration is a Configuration built from explicit values, standing in for the parsed +// token.selector configuration. +type fakeConfiguration struct { + enabled bool + rate float64 + burst int + maxBuckets int +} + +func (c *fakeConfiguration) IsRateLimitEnabled() bool { return c.enabled } +func (c *fakeConfiguration) GetRateLimit() float64 { return c.rate } +func (c *fakeConfiguration) GetRateLimitBurst() int { return c.burst } +func (c *fakeConfiguration) GetRateLimitMaxBuckets() int { return c.maxBuckets } + +// TestOptions_DisabledByDefault verifies that without options and without configuration there is no +// limiter at all. +func TestOptions_DisabledByDefault(t *testing.T) { + assert.Nil(t, CompileOptions().Limiter(&fakeConfiguration{})) + assert.Nil(t, CompileOptions().Limiter(nil)) +} + +// TestOptions_FromConfiguration verifies an enabled configuration produces a limiter with the +// configured rate and burst. +func TestOptions_FromConfiguration(t *testing.T) { + limiter := CompileOptions().Limiter(&fakeConfiguration{enabled: true, rate: 2, burst: 1}) + require.NotNil(t, limiter) + + ctx := context.Background() + require.NoError(t, limiter.Allow(ctx, testScope, "alice")) + require.ErrorIs(t, limiter.Allow(ctx, testScope, "alice"), token.SelectorRateLimited) +} + +// TestOptions_WithLimiterOverridesConfiguration verifies an explicitly supplied limiter wins over +// the configuration. +func TestOptions_WithLimiterOverridesConfiguration(t *testing.T) { + supplied := &countingLimiter{} + + limiter := CompileOptions(WithLimiter(supplied)).Limiter(&fakeConfiguration{enabled: true, rate: 100}) + require.Same(t, supplied, limiter) +} + +// TestOptions_WithNilLimiterDisables verifies WithLimiter(nil) switches rate limiting off even when +// the configuration enables it. +func TestOptions_WithNilLimiterDisables(t *testing.T) { + limiter := CompileOptions(WithLimiter(nil)).Limiter(&fakeConfiguration{enabled: true, rate: 100}) + assert.Nil(t, limiter) +} + +// TestOptions_WithDefaultLimiter verifies the built-in limiter can be enabled from code alone, with +// the documented default burst of 200 requests. +func TestOptions_WithDefaultLimiter(t *testing.T) { + limiter := CompileOptions(WithDefaultLimiter()).Limiter(&fakeConfiguration{}) + require.NotNil(t, limiter) + + bucketLimiter, ok := limiter.(*BucketLimiter) + require.True(t, ok) + assert.InDelta(t, DefaultRate, bucketLimiter.rate, 0) + assert.InDelta(t, 200.0, bucketLimiter.burst, 0) +} + +// TestOptions_LastOptionWins verifies options are applied in order. +func TestOptions_LastOptionWins(t *testing.T) { + supplied := &countingLimiter{} + + assert.Nil(t, CompileOptions(WithLimiter(supplied), WithLimiter(nil)).Limiter(nil)) + assert.Same(t, supplied, CompileOptions(WithLimiter(nil), WithLimiter(supplied)).Limiter(nil)) +} + +// TestOptions_NilOptionIsIgnored verifies a nil Option does not panic, so a caller assembling +// options conditionally does not have to filter them. +func TestOptions_NilOptionIsIgnored(t *testing.T) { + assert.Nil(t, CompileOptions(nil).Limiter(&fakeConfiguration{})) +} diff --git a/token/services/selector/sherdlock/ratelimit_test.go b/token/services/selector/sherdlock/ratelimit_test.go new file mode 100644 index 0000000000..1298082812 --- /dev/null +++ b/token/services/selector/sherdlock/ratelimit_test.go @@ -0,0 +1,180 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package sherdlock_test + +import ( + "context" + "fmt" + "testing" + + "github.com/LFDT-Panurus/panurus/token" + "github.com/LFDT-Panurus/panurus/token/driver" + drivermock "github.com/LFDT-Panurus/panurus/token/driver/mock" + "github.com/LFDT-Panurus/panurus/token/services/selector/ratelimit" + "github.com/LFDT-Panurus/panurus/token/services/selector/sherdlock" + "github.com/LFDT-Panurus/panurus/token/services/selector/sherdlock/mocks" + token2 "github.com/LFDT-Panurus/panurus/token/token" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newFetcherOf returns a fetcher that hands out a fresh iterator over a single token of the given +// quantity on every call, so repeated selections all see the same funds. +func newFetcherOf(quantity string) *mocks.FakeTokenFetcher { + fetcher := &mocks.FakeTokenFetcher{} + fetcher.UnspentTokensIteratorByStub = func(context.Context, string, token2.Type) (sherdlock.Iterator[*token2.UnspentTokenInWallet], error) { + it := &mocks.FakeIterator[*token2.UnspentTokenInWallet]{} + it.NextReturnsOnCall(0, &token2.UnspentTokenInWallet{ + Id: token2.ID{TxId: "tx1", Index: 0}, + Type: "ABC", + Quantity: quantity, + }, nil) + it.NextReturnsOnCall(1, nil, nil) + + return it, nil + } + + return fetcher +} + +// TestRateLimitedManager verifies that the built-in limiter, wrapped around a real sherdlock +// manager, throttles per wallet without leaking locks: the denied selection never reaches the +// locker, so there is nothing left locked, and releasing tokens keeps working while throttled. +func TestRateLimitedManager(t *testing.T) { + _, metrics := setupMetricsMocks() + fetcher := newFetcherOf("100") + locker := &mocks.FakeLocker{} + ctx := t.Context() + + mgr := sherdlock.NewManager(fetcher, locker, 64, 0, 0, 0, 0, metrics) + t.Cleanup(func() { require.NoError(t, mgr.Stop()) }) + + // A burst of one request, and a rate slow enough that nothing refills during the test. + limited := ratelimit.Decorate(mgr, ratelimit.New(ratelimit.Config{Rate: 0.001, Burst: 1}), "net1,ch1,ns1") + + selector, err := limited.NewSelector("tx1") + require.NoError(t, err) + + // The first selection spends the wallet's single request and locks its token. + tokens, sum, err := selector.Select(ctx, &unitTestMockOwnerFilter{id: "alice"}, "50", "ABC") + require.NoError(t, err) + assert.Len(t, tokens, 1) + assert.Equal(t, "100", sum.Decimal()) + locksSoFar := locker.LockCallCount() + assert.Equal(t, 1, locksSoFar) + + // The second one is denied before any token is even looked at. + tokens, sum, err = selector.Select(ctx, &unitTestMockOwnerFilter{id: "alice"}, "50", "ABC") + require.ErrorIs(t, err, token.SelectorRateLimited) + assert.Nil(t, tokens) + assert.Nil(t, sum) + assert.Equal(t, locksSoFar, locker.LockCallCount(), "a denied selection must not lock anything") + assert.Equal(t, 0, locker.UnlockByTxIDCallCount(), "a denied selection has nothing to release") + + // Another wallet is unaffected. + _, _, err = selector.Select(ctx, &unitTestMockOwnerFilter{id: "bob"}, "50", "ABC") + require.NoError(t, err) + + // Releasing the tokens of a throttled wallet must always be possible. + require.NoError(t, limited.Unlock(ctx, "tx1")) + assert.Equal(t, 1, locker.UnlockByTxIDCallCount()) + require.NoError(t, limited.Close("tx1")) +} + +// TestServiceRateLimiting verifies the service wiring: no rate limiting by default, and a limiter +// installed through the options that throttles the managers the service hands out. +func TestServiceRateLimiting(t *testing.T) { + t.Run("DisabledByDefault", func(t *testing.T) { + svc, tms := newServiceUnderTest(t) + + mgr, err := svc.SelectorManager(tms) + require.NoError(t, err) + // Without configuration or options the manager is handed out undecorated. + assert.IsType(t, &sherdlock.Manager{}, mgr) + }) + + t.Run("WithLimiter", func(t *testing.T) { + limiter := ratelimit.New(ratelimit.Config{Rate: 0.001, Burst: 1}) + svc, tms := newServiceUnderTest(t, ratelimit.WithLimiter(limiter)) + + mgr, err := svc.SelectorManager(tms) + require.NoError(t, err) + assert.NotEqual(t, "*sherdlock.Manager", fmt.Sprintf("%T", mgr), "the manager must be decorated by the limiter") + + selector, err := mgr.NewSelector("tx1") + require.NoError(t, err) + + // Spend the wallet's single request, then check the denial comes from the limiter, + // before the selector touches the (absent) lock store. + require.NoError(t, limiter.Allow(t.Context(), tms.ID().String(), "alice")) + _, _, err = selector.Select(t.Context(), &unitTestMockOwnerFilter{id: "alice"}, "50", "ABC") + require.ErrorIs(t, err, token.SelectorRateLimited) + }) + + t.Run("WithNilLimiter", func(t *testing.T) { + svc, tms := newServiceUnderTest(t, ratelimit.WithLimiter(nil)) + + mgr, err := svc.SelectorManager(tms) + require.NoError(t, err) + assert.IsType(t, &sherdlock.Manager{}, mgr) + }) + + t.Run("ShutdownKeepsTheLimiter", func(t *testing.T) { + limiter := ratelimit.New(ratelimit.Config{Rate: 0.001, Burst: 1}) + svc, tms := newServiceUnderTest(t, ratelimit.WithLimiter(limiter)) + + _, err := svc.SelectorManager(tms) + require.NoError(t, err) + require.NoError(t, limiter.Allow(t.Context(), tms.ID().String(), "alice")) + + // Shutdown also runs on public-parameter reloads, so it must not hand the wallet a + // fresh allowance. + svc.Shutdown() + require.ErrorIs(t, limiter.Allow(t.Context(), tms.ID().String(), "alice"), token.SelectorRateLimited) + }) +} + +// newServiceUnderTest returns a sherdlock service backed by mocks, together with the management +// service to ask it for a manager. +func newServiceUnderTest(t *testing.T, opts ...ratelimit.Option) (*sherdlock.SelectorService, *token.ManagementService) { + t.Helper() + + fetcherProvider := &mocks.FakeFetcherProvider{} + fetcherProvider.GetFetcherReturns(newFetcherOf("100"), nil) + lockStoreManager := &mocks.FakeTokenLockStoreServiceManager{} + lockStoreManager.StoreServiceByTMSIdReturns(nil, nil) + metricsProvider, _ := setupMetricsMocks() + + svc := sherdlock.NewService(fetcherProvider, lockStoreManager, &mocks.FakeConfigProvider{}, metricsProvider, opts...) + t.Cleanup(svc.Shutdown) + + driverTMS := &drivermock.TokenManagerService{} + ppm := &drivermock.PublicParamsManager{} + pp := &drivermock.PublicParameters{} + pp.PrecisionReturns(64) + ppm.PublicParametersReturns(pp) + driverTMS.PublicParamsManagerReturns(ppm) + + tms, err := token.NewManagementService( + token.TMSID{Network: "n1", Channel: "c1", Namespace: "ns1"}, + driverTMS, + nil, + &rateLimitMockVaultProvider{}, + nil, + nil, + ) + require.NoError(t, err) + + return svc, tms +} + +// rateLimitMockVaultProvider is the minimal VaultProvider NewManagementService needs. +type rateLimitMockVaultProvider struct{} + +func (v *rateLimitMockVaultProvider) Vault(network, channel, namespace string) (driver.Vault, error) { + return &drivermock.Vault{}, nil +} diff --git a/token/services/selector/sherdlock/service.go b/token/services/selector/sherdlock/service.go index 841256da50..be0be3dbb1 100644 --- a/token/services/selector/sherdlock/service.go +++ b/token/services/selector/sherdlock/service.go @@ -13,6 +13,7 @@ import ( "github.com/LFDT-Panurus/panurus/token" "github.com/LFDT-Panurus/panurus/token/core/common/metrics" "github.com/LFDT-Panurus/panurus/token/services/selector/config" + "github.com/LFDT-Panurus/panurus/token/services/selector/ratelimit" "github.com/LFDT-Panurus/panurus/token/services/storage/tokenlockdb" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" lazy2 "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/lazy" @@ -24,15 +25,21 @@ type SelectorService struct { managers []*Manager } +// NewService returns a SelectorService for the sherdlock driver. +// +// By default, selection is not rate limited. Passing ratelimit options, or enabling the +// token.selector.rateLimit* configuration keys, meters every selection request per wallet. func NewService( fetcherProvider FetcherProvider, tokenLockStoreServiceManager tokenlockdb.StoreServiceManager, c ConfigProvider, metricsProvider metrics.Provider, + opts ...ratelimit.Option, ) *SelectorService { cfg, err := config.New(c) if err != nil { logger.Errorf("error getting selector config, using defaults. %s", err.Error()) + cfg = &config.Config{} } svc := &SelectorService{} @@ -44,8 +51,12 @@ func NewService( leaseExpiry: cfg.GetLeaseExpiry(), leaseCleanupTickPeriod: cfg.GetLeaseCleanupTickPeriod(), metrics: NewMetrics(metricsProvider), + limiter: ratelimit.CompileOptions(opts...).Limiter(cfg), onCreate: svc.trackManager, } + if loader.limiter != nil { + logger.Infof("per-wallet token selection rate limiting is enabled") + } svc.managerLazyCache = lazy2.NewProviderWithKeyMapper(key, loader.load) return svc @@ -60,6 +71,13 @@ func (s *SelectorService) SelectorManager(tms *token.ManagementService) (token.S } // Shutdown stops all background goroutines for every manager created by this service. +// +// It deliberately leaves the rate limiter alone. Shutdown also runs on routine public-parameter +// reloads (see token.ManagementServiceProvider.Update), after which the service keeps serving +// managers: resetting the wallet allowances there would let a throttled client wash out its debt +// by triggering a reload, and a limiter supplied through ratelimit.WithLimiter belongs to the +// caller in the first place. The built-in limiter runs no goroutines and prunes its own buckets, +// so there is nothing to leak. func (s *SelectorService) Shutdown() { s.mu.Lock() managers := s.managers @@ -94,7 +112,10 @@ type loader struct { leaseExpiry time.Duration leaseCleanupTickPeriod time.Duration metrics *Metrics - onCreate func(*Manager) + // limiter meters selection requests per wallet. It is nil when rate limiting is + // disabled, which is the default, and is shared by every manager the loader builds. + limiter ratelimit.Limiter + onCreate func(*Manager) } func (s *loader) load(tms *token.ManagementService) (token.SelectorManager, error) { @@ -129,7 +150,8 @@ func (s *loader) loadTMS(tms TMS) (token.SelectorManager, error) { s.onCreate(mgr) } - return mgr, nil + // Decorate returns mgr unchanged when no limiter is configured. + return ratelimit.Decorate(mgr, s.limiter, tms.ID().String()), nil } func key(tms *token.ManagementService) string { diff --git a/token/services/selector/simple/ratelimit_test.go b/token/services/selector/simple/ratelimit_test.go new file mode 100644 index 0000000000..fbd906d1aa --- /dev/null +++ b/token/services/selector/simple/ratelimit_test.go @@ -0,0 +1,111 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package simple + +import ( + "context" + "testing" + + "github.com/LFDT-Panurus/panurus/token" + "github.com/LFDT-Panurus/panurus/token/services/selector/ratelimit" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestRateLimitedManager verifies that the built-in limiter, wrapped around a real simple manager, +// throttles per wallet without leaking locks: the denied selection never reaches the locker, so +// nothing is left locked and nothing has to be released. +func TestRateLimitedManager(t *testing.T) { + locker := &recordingLocker{lockFailAfter: 100} // every lock succeeds + qs := &mockQueryService{tokens: makeTokens(2, "USD", -1)} + ctx := context.Background() + + mgr := NewManager(locker, func() QueryService { return qs }, 1, 0, false, precision) + // A burst of one request, and a rate slow enough that nothing refills during the test. + limited := ratelimit.Decorate(mgr, ratelimit.New(ratelimit.Config{Rate: 0.001, Burst: 1}), "n1,c1,ns1") + + selector, err := limited.NewSelector("testTx") + require.NoError(t, err) + + // The first selection spends the wallet's single request and locks a token. + ids, sum, err := selector.Select(ctx, &ownerFilter{id: "wallet1"}, "0x1", "USD") + require.NoError(t, err) + assert.Len(t, ids, 1) + assert.NotNil(t, sum) + locksSoFar := locker.calls + assert.Positive(t, locksSoFar) + + // The second one is denied before any token is even looked at. + ids, sum, err = selector.Select(ctx, &ownerFilter{id: "wallet1"}, "0x1", "USD") + require.ErrorIs(t, err, token.SelectorRateLimited) + assert.Nil(t, ids) + assert.Nil(t, sum) + assert.Equal(t, locksSoFar, locker.calls, "a denied selection must not lock anything") + assert.Empty(t, locker.totalUnlocked(), "a denied selection has nothing to release") + + // Another wallet is unaffected. + _, _, err = selector.Select(ctx, &ownerFilter{id: "wallet2"}, "0x1", "USD") + require.NoError(t, err) + + // Releasing the tokens of a throttled wallet must always be possible. + require.NoError(t, limited.Unlock(ctx, "testTx")) + require.NoError(t, limited.Close("testTx")) +} + +// TestLoaderRateLimiting verifies the service wiring: managers are handed out undecorated unless a +// limiter is configured or supplied. +func TestLoaderRateLimiting(t *testing.T) { + tests := []struct { + name string + opts []ratelimit.Option + decorated bool + }{ + {name: "disabled by default", opts: nil, decorated: false}, + {name: "explicitly disabled", opts: []ratelimit.Option{ratelimit.WithLimiter(nil)}, decorated: false}, + {name: "default limiter", opts: []ratelimit.Option{ratelimit.WithDefaultLimiter()}, decorated: true}, + { + name: "custom limiter", + opts: []ratelimit.Option{ratelimit.WithLimiter(ratelimit.New(ratelimit.Config{Rate: 1, Burst: 1}))}, + decorated: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + locker := &recordingLocker{lockFailAfter: 100} + qs := &mockQueryService{tokens: makeTokens(1, "USD", -1)} + l := &loader{ + lockerProvider: &fixedLockerProvider{locker: locker}, + numRetries: 1, + requestCertification: false, + limiter: ratelimit.CompileOptions(tt.opts...).Limiter(&disabledRateLimitConfig{}), + } + + mgr := l.newManager(locker, qs, precision, "n1,c1,ns1") + _, plain := mgr.(*Manager) + assert.Equal(t, tt.decorated, !plain, "manager decorated: %t", !plain) + }) + } +} + +// fixedLockerProvider hands out a fixed Locker. +type fixedLockerProvider struct { + locker Locker +} + +func (p *fixedLockerProvider) New(_, _, _ string) (Locker, error) { + return p.locker, nil +} + +// disabledRateLimitConfig is a ratelimit.Configuration with rate limiting off, standing in for the +// default token.selector configuration. +type disabledRateLimitConfig struct{} + +func (c *disabledRateLimitConfig) IsRateLimitEnabled() bool { return false } +func (c *disabledRateLimitConfig) GetRateLimit() float64 { return 0 } +func (c *disabledRateLimitConfig) GetRateLimitBurst() int { return 0 } +func (c *disabledRateLimitConfig) GetRateLimitMaxBuckets() int { return 0 } diff --git a/token/services/selector/simple/service.go b/token/services/selector/simple/service.go index d3bdbfd4f4..8ff50240f8 100644 --- a/token/services/selector/simple/service.go +++ b/token/services/selector/simple/service.go @@ -15,6 +15,7 @@ import ( "github.com/LFDT-Panurus/panurus/token/driver" "github.com/LFDT-Panurus/panurus/token/services/logging" "github.com/LFDT-Panurus/panurus/token/services/selector/config" + "github.com/LFDT-Panurus/panurus/token/services/selector/ratelimit" token2 "github.com/LFDT-Panurus/panurus/token/token" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/lazy" @@ -41,10 +42,15 @@ type SelectorService struct { lockers []stoppable } -func NewService(lockerProvider LockerProvider, c ConfigProvider) *SelectorService { +// NewService returns a SelectorService for the simple driver. +// +// By default, selection is not rate limited. Passing ratelimit options, or enabling the +// token.selector.rateLimit* configuration keys, meters every selection request per wallet. +func NewService(lockerProvider LockerProvider, c ConfigProvider, opts ...ratelimit.Option) *SelectorService { cfg, err := config.New(c) if err != nil { logger.Errorf("error getting selector config, using defaults. %s", err.Error()) + cfg = &config.Config{} } svc := &SelectorService{} @@ -53,8 +59,12 @@ func NewService(lockerProvider LockerProvider, c ConfigProvider) *SelectorServic numRetries: cfg.GetNumRetries(), retryInterval: cfg.GetRetryInterval(), requestCertification: true, + limiter: ratelimit.CompileOptions(opts...).Limiter(cfg), onLockerCreated: svc.trackLocker, } + if loader.limiter != nil { + logger.Infof("per-wallet token selection rate limiting is enabled") + } svc.managerLazyCache = lazy.NewProviderWithKeyMapper(key, loader.load) return svc @@ -69,6 +79,13 @@ func (s *SelectorService) SelectorManager(tms *token.ManagementService) (token.S } // Shutdown stops all background goroutines for every locker created by this service. +// +// It deliberately leaves the rate limiter alone. Shutdown also runs on routine public-parameter +// reloads (see token.ManagementServiceProvider.Update), after which the service keeps serving +// managers: resetting the wallet allowances there would let a throttled client wash out its debt +// by triggering a reload, and a limiter supplied through ratelimit.WithLimiter belongs to the +// caller in the first place. The built-in limiter runs no goroutines and prunes its own buckets, +// so there is nothing to leak. func (s *SelectorService) Shutdown() { s.mu.Lock() lockers := s.lockers @@ -117,7 +134,10 @@ type loader struct { numRetries int retryInterval time.Duration requestCertification bool - onLockerCreated func(Locker) + // limiter meters selection requests per wallet. It is nil when rate limiting is + // disabled, which is the default, and is shared by every manager the loader builds. + limiter ratelimit.Limiter + onLockerCreated func(Locker) } func (s *loader) load(tms *token.ManagementService) (token.SelectorManager, error) { @@ -135,14 +155,24 @@ func (s *loader) load(tms *token.ManagementService) (token.SelectorManager, erro locker: locker, } - return NewManager( + return s.newManager(locker, qe, tms.PublicParametersManager().PublicParameters().Precision(), tms.ID().String()), nil +} + +// newManager builds the manager for one TMS and, when rate limiting is enabled, wraps it so that +// every selection request is metered against the allowance of the wallet it selects for. scope is +// the TMS id, which keeps the allowances of one network or namespace separate from the others. +func (s *loader) newManager(locker Locker, qs QueryService, precision uint64, scope string) token.SelectorManager { + mgr := NewManager( locker, - func() QueryService { return qe }, + func() QueryService { return qs }, s.numRetries, s.retryInterval, s.requestCertification, - tms.PublicParametersManager().PublicParameters().Precision(), - ), nil + precision, + ) + + // Decorate returns mgr unchanged when no limiter is configured, which is the default. + return ratelimit.Decorate(mgr, s.limiter, scope) } func key(tms *token.ManagementService) string {