mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-19 11:09:26 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
parent
9bf7e57f92
commit
9e2a781507
@ -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()
|
||||
|
||||
@ -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()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user