feat: split rate limit manager into config, quota source, and enforcement modules

Decomposes logos_delivery/messaging/rate_limit_manager/ into three
modules with one responsibility each:

- rate_limit_config: the configuration vocabulary (RateLimitConfig,
  RateLimitError, defaults, isEnforcing). The API conf layer now
  imports this alone instead of the enforcement engine.
- quota_source: the RLN seam. A single QuotaProvider callback returns
  EpochQuota (epoch index + user message limit) so the two are read
  atomically and a read cannot straddle an epoch boundary. Returns
  none when RLN is unavailable, selecting the wall-clock fallback. The
  callback shape keeps the manager free of any dependency on the Waku
  kernel; the RLN-backed provider is built one layer up.
- rate_limit_manager: enforcement only; re-exports the other two so
  existing single-import call sites are unchanged.

Config field names now mirror RLN exactly, per the requirement that the
rate limit share RLN Relay's format: epochSizeSec (was epochPeriodSec)
and userMessageLimit (was messagesPerEpoch), both uint64 to match
RlnConf. Renaming is contained to this branch: the config type moved
into the new rate_limit_config module here.

admit() now works in epoch-index terms: the epoch comes from the
provider when set (RLN's calcEpoch value), else from an absolute
wall-clock window (unixTime div epochSizeSec — absolute rather than
anchored at first use, matching RLN's derivation). The effective limit
is min(config.userMessageLimit, RLN's) — RLN can only tighten the
configured limit, since exceeding it would fail at proof generation
once the epoch's message ids are exhausted.

With no provider wired (this commit), behaviour is the wall-clock
window as before; RLN-backed provider wiring follows separately.

Tests rewritten against injected fake providers: limit boundary, epoch
rollover without sleeps, RLN-clamps-config, config-tightens-below-RLN,
and the wall-clock fallback (7 tests).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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.

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

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

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

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