From 9e2a7815072f92af047d1a7775e8711ff9220153 Mon Sep 17 00:00:00 2001 From: stubbsta Date: Fri, 10 Jul 2026 14:34:54 +0200 Subject: [PATCH] feat: enforce per-epoch budget in RateLimitManager.admit Replaces the pass-through skeleton with a lazily rolled fixed window: admit() charges one message against the current epoch, resets the counter once epochPeriodSec has elapsed, and rejects with OverBudget when messagesPerEpoch is exhausted. Disabled or non-positive configurations admit everything, so the default-constructed MessagingClientConf (enabled = false) keeps today's behaviour. Parking of over-budget messages stays with the SendService scheduler (NextRoundRetry); the manager only answers whether one more transmission fits. The queue / dequeueReady stubs that anticipated manager-side parking are removed accordingly. Extends tests/messaging/test_rate_limit_manager.nim with budget boundary, epoch rollover, resetEpoch, and degenerate-config cases, replacing the enabled-pass-through placeholder test. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../rate_limit_manager/rate_limit_manager.nim | 48 +++++++++++-------- tests/messaging/test_rate_limit_manager.nim | 44 ++++++++++++++--- 2 files changed, 64 insertions(+), 28 deletions(-) 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 7ee197d8a..67aceae98 100644 --- a/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim @@ -1,12 +1,13 @@ ## Rate Limit Manager for the Messaging API. ## ## Tracks messages sent per RLN epoch and rejects admission when the -## limit is approached, ensuring RLN compliance on enforcing relays. +## budget is exhausted, ensuring RLN compliance on enforcing relays. ## -## For the skeleton this is a pass-through: every call is admitted. -## Real per-epoch budgeting will use `queue`, `currentEpochStart`, -## `sentInCurrentEpoch`, and `resetEpoch` to park messages and admit -## them as the epoch rolls over. +## 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. ## ## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html @@ -24,7 +25,6 @@ type RateLimitManager* = ref object config*: RateLimitConfig - queue*: seq[seq[byte]] currentEpochStart*: Time sentInCurrentEpoch*: int @@ -33,22 +33,28 @@ const DefaultMessagesPerEpoch* = 1 proc new*(T: type RateLimitManager, config: RateLimitConfig): T = - return - T(config: config, queue: @[], currentEpochStart: getTime(), sentInCurrentEpoch: 0) - -proc admit*( - self: RateLimitManager, msg: seq[byte] -): Future[Result[void, RateLimitError]] {.async: (raises: []).} = - ## Skeleton behaviour: admits immediately. Real per-epoch budgeting - ## will consult `config`, `sentInCurrentEpoch`, and the elapsed - ## `epochPeriodSec` window before admitting or parking `msg`. - return ok() - -proc dequeueReady*(self: RateLimitManager): seq[seq[byte]] = - ## Returns the set of queued messages that may be dispatched now - ## without exceeding the configured rate limit. - discard + return T(config: config, currentEpochStart: getTime(), sentInCurrentEpoch: 0) proc resetEpoch*(self: RateLimitManager) = self.currentEpochStart = getTime() self.sentInCurrentEpoch = 0 + +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: + return ok() + + if getTime() - self.currentEpochStart >= + initDuration(seconds = self.config.epochPeriodSec): + self.resetEpoch() + + if self.sentInCurrentEpoch >= self.config.messagesPerEpoch: + return err(RateLimitError.OverBudget) + + inc self.sentInCurrentEpoch + return ok() diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim index 895fcb49d..bd259a19f 100644 --- a/tests/messaging/test_rate_limit_manager.nim +++ b/tests/messaging/test_rate_limit_manager.nim @@ -13,14 +13,44 @@ suite "RateLimitManager - admission": let res = await rl.admit("payload".toBytes()) check res.isOk() - asyncTest "admit is a pass-through in the skeleton even when enabled": - ## Documents the current skeleton behaviour: per-epoch enforcement is - ## not wired yet, so every call is admitted regardless of the - ## configured budget. This test flips to red as soon as real - ## enforcement lands, at which point it should be replaced with - ## budget-boundary assertions. + asyncTest "admits up to the budget then rejects with OverBudget": + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 3) + ) + for i in 0 ..< 3: + check (await rl.admit(("msg" & $i).toBytes())).isOk() + let res = await rl.admit("over".toBytes()) + check: + res.isErr() + res.error == RateLimitError.OverBudget + + asyncTest "budget frees when the epoch rolls over": + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 1, messagesPerEpoch: 1) + ) + check (await rl.admit("first".toBytes())).isOk() + check (await rl.admit("second".toBytes())).isErr() + await sleepAsync(1100.milliseconds) + check (await rl.admit("third".toBytes())).isOk() + check (await rl.admit("fourth".toBytes())).isErr() + + asyncTest "resetEpoch forces a fresh budget": let rl = RateLimitManager.new( RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1) ) check (await rl.admit("first".toBytes())).isOk() - check (await rl.admit("second".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()