From cdafbba41a98ee1a752784ec7e86c50990d1630c Mon Sep 17 00:00:00 2001 From: stubbsta Date: Fri, 17 Jul 2026 13:15:54 +0200 Subject: [PATCH] feat: split rate limit manager into config, quota source, and enforcement modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Decomposes logos_delivery/messaging/rate_limit_manager/ into three modules with one responsibility each: - rate_limit_config: the configuration vocabulary (RateLimitConfig, RateLimitError, defaults, isEnforcing). The API conf layer now imports this alone instead of the enforcement engine. - quota_source: the RLN seam. A single QuotaProvider callback returns EpochQuota (epoch index + user message limit) so the two are read atomically and a read cannot straddle an epoch boundary. Returns none when RLN is unavailable, selecting the wall-clock fallback. The callback shape keeps the manager free of any dependency on the Waku kernel; the RLN-backed provider is built one layer up. - rate_limit_manager: enforcement only; re-exports the other two so existing single-import call sites are unchanged. Config field names now mirror RLN exactly, per the requirement that the rate limit share RLN Relay's format: epochSizeSec (was epochPeriodSec) and userMessageLimit (was messagesPerEpoch), both uint64 to match RlnConf. Renaming is contained to this branch: the config type moved into the new rate_limit_config module here. admit() now works in epoch-index terms: the epoch comes from the provider when set (RLN's calcEpoch value), else from an absolute wall-clock window (unixTime div epochSizeSec — absolute rather than anchored at first use, matching RLN's derivation). The effective limit is min(config.userMessageLimit, RLN's) — RLN can only tighten the configured limit, since exceeding it would fail at proof generation once the epoch's message ids are exhausted. With no provider wired (this commit), behaviour is the wall-clock window as before; RLN-backed provider wiring follows separately. Tests rewritten against injected fake providers: limit boundary, epoch rollover without sleeps, RLN-clamps-config, config-tightens-below-RLN, and the wall-clock fallback (7 tests). Co-Authored-By: Claude Opus 4.8 (1M context) --- logos_delivery/api/conf/messaging_conf.nim | 6 +- .../rate_limit_manager/quota_source.nim | 38 ++++++++ .../rate_limit_manager/rate_limit_config.nim | 35 +++++++ .../rate_limit_manager/rate_limit_manager.nim | 95 +++++++++++-------- tests/messaging/test_rate_limit_manager.nim | 75 ++++++++++----- 5 files changed, 187 insertions(+), 62 deletions(-) create mode 100644 logos_delivery/messaging/rate_limit_manager/quota_source.nim create mode 100644 logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index 3ec35481a..af406ea40 100644 --- a/logos_delivery/api/conf/messaging_conf.nim +++ b/logos_delivery/api/conf/messaging_conf.nim @@ -4,9 +4,9 @@ import results, libp2p/crypto/crypto import logos_delivery/api/conf/kernel_conf import logos_delivery/waku/common/logging import logos_delivery/waku/factory/networks_config -import logos_delivery/messaging/rate_limit_manager/rate_limit_manager +import logos_delivery/messaging/rate_limit_manager/rate_limit_config -export kernel_conf, rate_limit_manager +export kernel_conf, rate_limit_config # `LogosDeliveryMode` and `EntryLayer` are defined at the leaf (`cli_args`) so # they can appear on `WakuNodeConf`; re-exported here via `kernel_conf`. @@ -53,7 +53,7 @@ type MessagingClientConf* = object ## P2P node private key (64-char hex): stable identity / peerId across restarts. rateLimit*: RateLimitConfig = RateLimitConfig( epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch - ) ## RLN-epoch transmission budget enforced by the send service. + ) ## Per-epoch message rate limit enforced by the send service. proc applyMode*(conf: var WakuNodeConf, mode: LogosDeliveryMode): ConfResult[void] = ## Sets the protocol flags implied by the mode. diff --git a/logos_delivery/messaging/rate_limit_manager/quota_source.nim b/logos_delivery/messaging/rate_limit_manager/quota_source.nim new file mode 100644 index 000000000..f46132a43 --- /dev/null +++ b/logos_delivery/messaging/rate_limit_manager/quota_source.nim @@ -0,0 +1,38 @@ +## Epoch + user-message-limit source for the rate limit manager. +## +## "Quota" is the tracking issues' word (#3838: "we should know our remaining +## quota") for what RLN expresses as an epoch's user message limit: each epoch +## grants `userMessageLimit` message ids (nonces), and the epoch rolling over +## refills the allowance. +## +## The manager rate-limits per epoch, but must not know *where* the epoch and +## the limit come from. With RLN mounted both are RLN's; without it, a +## wall-clock window and the configured limit stand in. Epoch and limit are +## read together, through one provider, so a read cannot straddle an epoch +## boundary and pair a fresh epoch with a stale limit. The provider is a +## callback so the manager stays free of any dependency on the `Waku` kernel — +## the RLN-backed provider is built one layer up, where the kernel handle is in +## scope. It returns `none` when RLN is not mounted, which is the signal to +## fall back to the wall clock. + +import std/[options, times] + +type + EpochQuota* = object + epochIndex*: uint64 + ## Current epoch as its numeric value (`timestamp div` the epoch length); + ## the kernel's `Epoch` type is the same value serialized to 32 bytes. + userMessageLimit*: uint64 + ## Message ids the epoch grants — the hard ceiling on admissions before + ## the epoch rolls over. + + QuotaProvider* = proc(): Option[EpochQuota] {.gcsafe, raises: [].} + ## The epoch and its user message limit, or `none` when RLN is not mounted. + +proc wallClockEpochIndex*(epochPeriodSec: uint64): uint64 = + ## Absolute wall-clock epoch: `unixTime div epochPeriodSec`. Absolute (not a + ## sliding window anchored at first use) so it matches RLN's `calcEpoch`, and + ## so two managers started at different moments agree on the boundary. + ## `epochPeriodSec` is assumed positive; the manager only calls this when + ## enforcing. + uint64(getTime().toUnix()) div epochPeriodSec diff --git a/logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim b/logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim new file mode 100644 index 000000000..1fc4f7300 --- /dev/null +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim @@ -0,0 +1,35 @@ +## Configuration for the messaging rate limit manager. +## +## Field names follow the RELIABLE-CHANNEL-API spec's `RateLimitConfig` +## (`enabled`, `epochPeriodSec`): this is a messaging-API config object, so it +## uses the API spec's vocabulary rather than RLN's internal `RlnConf` names. +## `messagesPerEpoch` is the local cap; the spec leaves the actual limit to RLN, +## which the manager honours by clamping the cap to RLN's user message limit. +## +## Kept separate from the enforcement engine so the API config layer can depend +## on the vocabulary (`RateLimitConfig`) without pulling in the manager and its +## RLN quota seam. + +type + RateLimitError* {.pure.} = enum + OverBudget + + RateLimitConfig* = object + enabled*: bool ## spec: rate limiting opt-in; SHOULD be true when RLN active + epochPeriodSec*: uint64 + ## Epoch length, in seconds (spec: the epoch size used by the RLN relay). + ## Only shapes the wall-clock fallback window; when the RLN quota source is + ## wired in it is ignored (the period is RLN's). + messagesPerEpoch*: uint64 + ## Local cap on messages admitted per epoch. When RLN is mounted the + ## effective limit is clamped to RLN's user message limit; config can only + ## tighten it, never widen it. + +const + DefaultEpochPeriodSec* = 600'u64 + DefaultMessagesPerEpoch* = 1'u64 + +func isEnforcing*(config: RateLimitConfig): bool = + ## Whether the config asks for actual rate limiting. A disabled or zeroed + ## configuration admits everything, so the manager can short-circuit. + config.enabled and config.epochPeriodSec > 0 and config.messagesPerEpoch > 0 diff --git a/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim index 67aceae98..d6fb14724 100644 --- a/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim @@ -1,59 +1,80 @@ ## Rate Limit Manager for the Messaging API. ## -## Tracks messages sent per RLN epoch and rejects admission when the -## budget is exhausted, ensuring RLN compliance on enforcing relays. +## Rate-limits message transmissions against the per-epoch user message limit +## and rejects admission once it is spent, keeping the node within the quota +## that RLN-enforcing relays check. Each admission mirrors one RLN message id +## (nonce) draw; the epoch rolling over refills the allowance. ## -## Budgeting is a lazily rolled fixed window: the first admission after -## `epochPeriodSec` has elapsed resets the counter. Parking and retrying -## of over-budget messages is owned by the send service scheduler; this -## module only answers whether one more transmission fits the current -## epoch's budget. +## The epoch and the limit come from a `QuotaProvider` when RLN is mounted +## (see `quota_source`); otherwise a wall-clock window and the configured limit +## stand in. Parking and retrying over-budget messages is the send service +## scheduler's job — this module only answers whether one more transmission +## fits the current epoch. ## ## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html -import std/times +import std/options import results, chronos -type - RateLimitError* {.pure.} = enum - OverBudget +import ./rate_limit_config, ./quota_source - RateLimitConfig* = object - enabled*: bool ## spec: rate limiting opt-in; SHOULD be true when RLN active - epochPeriodSec*: int - messagesPerEpoch*: int +export rate_limit_config, quota_source - RateLimitManager* = ref object - config*: RateLimitConfig - currentEpochStart*: Time - sentInCurrentEpoch*: int +type RateLimitManager* = ref object + config*: RateLimitConfig + quotaProvider: QuotaProvider + ## Supplies the RLN epoch + user message limit. Nil, or a call returning + ## `none`, selects the wall-clock fallback. Queried per admission so a node + ## whose RLN mounts after construction upgrades automatically. + currentEpochIndex*: uint64 + sentInCurrentEpoch*: uint64 -const - DefaultEpochPeriodSec* = 600 - DefaultMessagesPerEpoch* = 1 +proc new*( + T: type RateLimitManager, + config: RateLimitConfig, + quotaProvider: QuotaProvider = nil, +): T = + return T( + config: config, + quotaProvider: quotaProvider, + currentEpochIndex: 0, + sentInCurrentEpoch: 0, + ) -proc new*(T: type RateLimitManager, config: RateLimitConfig): T = - return T(config: config, currentEpochStart: getTime(), sentInCurrentEpoch: 0) - -proc resetEpoch*(self: RateLimitManager) = - self.currentEpochStart = getTime() - self.sentInCurrentEpoch = 0 +proc currentQuota(self: RateLimitManager): Option[EpochQuota] = + if self.quotaProvider.isNil(): + return none(EpochQuota) + return self.quotaProvider() proc admit*( self: RateLimitManager, msg: seq[byte] ): Future[Result[void, RateLimitError]] {.async: (raises: []).} = - ## Charges one message against the current epoch's budget, rolling the - ## window first when `epochPeriodSec` has elapsed. A disabled or - ## non-positive configuration admits everything. - if not self.config.enabled or self.config.epochPeriodSec <= 0 or - self.config.messagesPerEpoch <= 0: + ## Charges one message against the current epoch's user message limit, + ## rolling the window first when the epoch has advanced. A disabled or zeroed + ## configuration admits everything. + if not self.config.isEnforcing(): return ok() - if getTime() - self.currentEpochStart >= - initDuration(seconds = self.config.epochPeriodSec): - self.resetEpoch() + let quota = self.currentQuota() - if self.sentInCurrentEpoch >= self.config.messagesPerEpoch: + let epochIndex = + if quota.isSome(): + quota.get().epochIndex + else: + wallClockEpochIndex(self.config.epochPeriodSec) + + # RLN can only tighten the configured cap, never widen it: exceeding RLN's + # user message limit would fail later at proof generation, once the epoch's + # message ids are exhausted. + var limit = self.config.messagesPerEpoch + if quota.isSome() and quota.get().userMessageLimit < limit: + limit = quota.get().userMessageLimit + + if epochIndex != self.currentEpochIndex: + self.currentEpochIndex = epochIndex + self.sentInCurrentEpoch = 0 + + if self.sentInCurrentEpoch >= limit: return err(RateLimitError.OverBudget) inc self.sentInCurrentEpoch diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim index bd259a19f..68908d482 100644 --- a/tests/messaging/test_rate_limit_manager.nim +++ b/tests/messaging/test_rate_limit_manager.nim @@ -1,21 +1,41 @@ {.used.} +import std/options import chronos, testutils/unittests, stew/byteutils import logos_delivery/messaging/rate_limit_manager/rate_limit_manager +proc fixedQuota(epochIndex, userMessageLimit: uint64): QuotaProvider = + ## A quota source pinned to one epoch — the epoch never rolls on its own, so + ## limit-boundary tests are deterministic without touching the wall clock. + return proc(): Option[EpochQuota] {.gcsafe, raises: [].} = + some(EpochQuota(epochIndex: epochIndex, userMessageLimit: userMessageLimit)) + suite "RateLimitManager - admission": asyncTest "admit is a pass-through when disabled": let rl = RateLimitManager.new( RateLimitConfig(enabled: false, epochPeriodSec: 600, messagesPerEpoch: 1) ) for _ in 0 ..< 10: - let res = await rl.admit("payload".toBytes()) - check res.isOk() + check (await rl.admit("payload".toBytes())).isOk() - asyncTest "admits up to the budget then rejects with OverBudget": + asyncTest "zeroed limit or epoch period is treated as disabled": + let zeroLimit = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 0) + ) + check (await zeroLimit.admit("a".toBytes())).isOk() + + let zeroEpoch = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 0, messagesPerEpoch: 1) + ) + check (await zeroEpoch.admit("a".toBytes())).isOk() + check (await zeroEpoch.admit("b".toBytes())).isOk() + + asyncTest "admits up to the message limit then rejects with OverBudget": + ## Fixed-epoch quota so the window cannot roll mid-test. let rl = RateLimitManager.new( - RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 3) + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 3), + fixedQuota(epochIndex = 42, userMessageLimit = 100), ) for i in 0 ..< 3: check (await rl.admit(("msg" & $i).toBytes())).isOk() @@ -24,33 +44,44 @@ suite "RateLimitManager - admission": res.isErr() res.error == RateLimitError.OverBudget - asyncTest "budget frees when the epoch rolls over": + asyncTest "allowance refills when the epoch rolls over": + ## Drive the roll through the provider — no sleeps, no flake. + var epoch = 1'u64 let rl = RateLimitManager.new( - RateLimitConfig(enabled: true, epochPeriodSec: 1, messagesPerEpoch: 1) + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1), + proc(): Option[EpochQuota] {.gcsafe, raises: [].} = + some(EpochQuota(epochIndex: epoch, userMessageLimit: 100)), ) check (await rl.admit("first".toBytes())).isOk() check (await rl.admit("second".toBytes())).isErr() - await sleepAsync(1100.milliseconds) + epoch = 2 check (await rl.admit("third".toBytes())).isOk() check (await rl.admit("fourth".toBytes())).isErr() - asyncTest "resetEpoch forces a fresh budget": + asyncTest "RLN user message limit clamps a looser configured cap": + ## config cap 5, RLN grants 2 — the third admission must reject, since + ## exceeding RLN's limit would fail at proof generation anyway. + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 5), + fixedQuota(epochIndex = 7, userMessageLimit = 2), + ) + check (await rl.admit("a".toBytes())).isOk() + check (await rl.admit("b".toBytes())).isOk() + check (await rl.admit("c".toBytes())).isErr() + + asyncTest "config cap can tighten below the RLN limit": + ## config cap 1, RLN grants 100 — the config cap wins. + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1), + fixedQuota(epochIndex = 7, userMessageLimit = 100), + ) + check (await rl.admit("a".toBytes())).isOk() + check (await rl.admit("b".toBytes())).isErr() + + asyncTest "falls back to the wall-clock window when no quota source is set": + ## No provider: rate limiting still enforces within a single wall-clock epoch. let rl = RateLimitManager.new( RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1) ) check (await rl.admit("first".toBytes())).isOk() check (await rl.admit("second".toBytes())).isErr() - rl.resetEpoch() - check (await rl.admit("third".toBytes())).isOk() - - asyncTest "non-positive budget or period is treated as disabled": - let zeroBudget = RateLimitManager.new( - RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 0) - ) - check (await zeroBudget.admit("a".toBytes())).isOk() - - let zeroPeriod = RateLimitManager.new( - RateLimitConfig(enabled: true, epochPeriodSec: 0, messagesPerEpoch: 1) - ) - check (await zeroPeriod.admit("a".toBytes())).isOk() - check (await zeroPeriod.admit("b".toBytes())).isOk()