From 861376cd3cc91cdc57dcc5191fe420df0a605948 Mon Sep 17 00:00:00 2001 From: Ivan FB Date: Fri, 17 Jul 2026 17:49:17 +0200 Subject: [PATCH] feat: enforce per-epoch budget in RateLimitManager.admit admit now meters each transmission against messagesPerEpoch and rolls the epoch window when epochPeriodSec elapses, instead of admitting every call. This is the client-side rate gate the SendService already calls into. Drop the unused queue field and dequeueReady: the SendService owns the pending work as DeliveryTasks and re-offers parked ones each loop tick, so the manager needs no queue of its own. admit keeps its signature so both call sites stay untouched. Co-Authored-By: Claude Opus 4.8 --- .../rate_limit_manager/rate_limit_manager.nim | 52 +++++++++++-------- tests/messaging/test_rate_limit_manager.nim | 23 +++++--- 2 files changed, 47 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..e18b71dc6 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. +## configured 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. +## `admit` meters each call against `messagesPerEpoch`, rolling the epoch +## window over once `epochPeriodSec` has elapsed since `currentEpochStart`. +## The caller (SendService) owns the pending work and re-offers parked +## tasks on the next service-loop tick, so the manager keeps no queue. ## ## 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,32 @@ 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 rollEpochIfElapsed(self: RateLimitManager) = + ## Opens a fresh budget window once the configured epoch period has + ## passed, so `sentInCurrentEpoch` counts only the current epoch. + if getTime() - self.currentEpochStart >= + initDuration(seconds = self.config.epochPeriodSec): + self.resetEpoch() + +proc admit*( + self: RateLimitManager, msg: seq[byte] +): Future[Result[void, RateLimitError]] {.async: (raises: []).} = + ## Meters one transmission against the per-epoch budget: rolls the epoch + ## window if elapsed, then admits and counts the send while under + ## `messagesPerEpoch`, or returns `OverBudget` so the caller can park it. + if not self.config.enabled: + return ok() + + self.rollEpochIfElapsed() + + if self.sentInCurrentEpoch >= self.config.messagesPerEpoch: + return err(RateLimitError.OverBudget) + + self.sentInCurrentEpoch.inc() + return ok() diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim index 895fcb49d..6a4cfc8ce 100644 --- a/tests/messaging/test_rate_limit_manager.nim +++ b/tests/messaging/test_rate_limit_manager.nim @@ -13,14 +13,23 @@ 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 "admit enforces the per-epoch budget when enabled": 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() + + let overBudget = await rl.admit("second".toBytes()) + check overBudget.isErr() + check overBudget.error == RateLimitError.OverBudget + + asyncTest "admit reopens the budget after the epoch rolls over": + 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() + + # Simulate the epoch window elapsing; the next call must be admitted again. + rl.resetEpoch() + check (await rl.admit("third".toBytes())).isOk()