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 <noreply@anthropic.com>

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()

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()