diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index 32ab9b235..844186057 100644 --- a/logos_delivery/api/conf/messaging_conf.nim +++ b/logos_delivery/api/conf/messaging_conf.nim @@ -6,9 +6,9 @@ import 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 type LogosDeliveryMode* {.pure.} = enum Edge # client-only node @@ -56,8 +56,8 @@ type MessagingClientConf* = object nodeKey* {.name: "nodekey".}: Option[crypto.PrivateKey] ## 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. + epochSizeSec: DefaultEpochSizeSec, userMessageLimit: DefaultUserMessageLimit + ) ## Per-epoch user message 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..763e7bc16 --- /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 epochSizeSec`); + ## 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*(epochSizeSec: uint64): uint64 = + ## Absolute wall-clock epoch: `unixTime div epochSizeSec`. 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. + ## `epochSizeSec` is assumed positive; the manager only calls this when + ## enforcing. + uint64(getTime().toUnix()) div epochSizeSec 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..f225ade4e --- /dev/null +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim @@ -0,0 +1,31 @@ +## Configuration for the messaging rate limit manager. +## +## Field names follow RLN terminology (`epochSizeSec`, `userMessageLimit` — see +## `RlnConf` and the RLN v2 spec) since they configure the local mirror of the +## same per-epoch limit RLN enforces on the network. +## +## 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 seam. + +type + RateLimitError* {.pure.} = enum + OverBudget + + RateLimitConfig* = object + enabled*: bool ## spec: rate limiting opt-in; SHOULD be true when RLN active + epochSizeSec*: uint64 + ## Epoch length, in seconds. Only shapes the wall-clock fallback window; + ## when the RLN quota source is wired in it is ignored (the size is RLN's). + userMessageLimit*: uint64 + ## Messages allowed per epoch. When RLN is mounted the effective limit is + ## clamped to RLN's own user message limit; config can only tighten it. + +const + DefaultEpochSizeSec* = 600'u64 + DefaultUserMessageLimit* = 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.epochSizeSec > 0 and config.userMessageLimit > 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..1ca57ad1a 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.epochSizeSec) + + # RLN can only tighten the configured limit, never widen it: exceeding RLN's + # limit would fail later at proof generation, once the epoch's message ids + # are exhausted. + var limit = self.config.userMessageLimit + 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..270a20137 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) + RateLimitConfig(enabled: false, epochSizeSec: 600, userMessageLimit: 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 size is treated as disabled": + let zeroLimit = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 0) + ) + check (await zeroLimit.admit("a".toBytes())).isOk() + + let zeroEpoch = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 0, userMessageLimit: 1) + ) + check (await zeroEpoch.admit("a".toBytes())).isOk() + check (await zeroEpoch.admit("b".toBytes())).isOk() + + asyncTest "admits up to the user 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, epochSizeSec: 600, userMessageLimit: 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, epochSizeSec: 600, userMessageLimit: 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 limit": + ## config allows 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: 1) + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 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 can tighten below the RLN limit": + ## config allows 1, RLN grants 100 — the config limit wins. + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 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, epochSizeSec: 600, userMessageLimit: 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()