feat: enforce the per-epoch rate limit at the send service (RLN-sourced) (#4062)

* 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>

* 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>

* fix: charge admission once per task and exempt budget-parked tasks from the reaper

Two defects the send-service seam had once admission actually enforces,
both fixed by a single DeliveryTask.firstAdmittedTime field:

- The retry loop gated admission on firstPropagatedTime.isNone(), so a
  task that failed to propagate (e.g. no peers) was re-admitted on every
  1s tick, re-charging the epoch budget and — with the shipped default of
  one message per epoch — starving all other traffic. Admission is now
  gated on firstAdmittedTime.isNone(): a task charges one slot / draws one
  nonce for its lifetime, and retries reuse it.

- reportTaskResult reaped any never-propagated task older than
  MaxTimeInCache (60s) measured from message creation, so an over-budget
  task parked for a 600s epoch was hard-failed with a misleading "Unable
  to send within retry time window" long before the epoch could roll
  (issue #4049). The reaper now runs only for admitted tasks and measures
  from admission, so a task waiting for epoch budget is exempt and gets a
  full delivery window once it is finally admitted.

The RLN-rejection branch in both processors resets firstAdmittedTime
along with clearing the proof, so the regenerated proof's fresh nonce is
re-admitted rather than sent uncharged.

The reap decision is extracted to DeliveryTask.isDeliveryTimedOut and
unit-tested (tests/messaging/test_delivery_task_reaping.nim); a
scheduler-level integration test of park-and-release is a follow-up.

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

* feat: source the rate limit manager's epoch and budget from RLN

Wires the quota seam to its producer, so enforcement tracks RLN rather
than only the wall clock.

- Waku.currentRlnEpochQuota (waku/api/publish) reads RLN's current epoch
  index and the epoch's user message limit together, returning none when
  RLN is not mounted (or its limit is unset).
- MessagingClient.new builds a QuotaProvider closure over that accessor
  and hands it to the RateLimitManager. The closure is late-binding: it
  queries the kernel on each admission, so a node whose RLN mounts after
  construction upgrades from the wall-clock fallback to RLN's epoch and
  limit automatically, with no reconstruction.

With this, admit() rolls its window on RLN's epoch and clamps the
configured cap to RLN's user message limit; without RLN it still falls
back to the absolute wall-clock window and the configured limit.

Also switches the quota seam from std/options to results `Opt`, matching
the kernel surface it now bridges (`groupManager.userMessageLimit` is
`Opt`) and the rest of the messaging layer post-Opt migration.

Tests: currentRlnEpochQuota is none unmounted and reports epoch + the
configured userMessageLimit when mounted (anvil-backed).

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

* test: cover admit-once and park/release at the send service scheduler

The rate limit manager's budget logic and the delivery-timeout reaper are
unit-covered, but the send service's use of them across the service loop was
not. Drive the scheduler a tick at a time against a scripted fake processor
and a fixed-epoch quota provider — no network, no sleeps — asserting:

- a task is charged against the budget exactly once however many rounds
  delivery takes (firstAdmittedTime guards re-admission);
- an over-budget task parks as NextRoundRetry without reaching the processor,
  then is admitted and delivered on the first tick after the epoch rolls.

Two testability seams keep this deterministic without a live relay: an
optional sendProcessor override on SendService.new injects the fake, and
trySendMessages is exported to drive one loop tick. The task is built
directly (like test_delivery_task_reaping) since DeliveryTask.new resolves
its shard through a broker provider only registered once the node starts.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs: note rateLimit config is settable only programmatically

MessagingClientConf.rateLimit cannot be set through the JSON config or a CLI
flag: it carries no name pragma, and RateLimitConfig is a nested object with
no parseCmdArg, so applyJsonFieldsToConf rejects it with "cannot be set via
JSON". Record the limitation and the fix path on the field.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* refactor: consolidate the admission gate into admitOnce

send() and the retry loop duplicated the admit-then-stamp-firstAdmittedTime
sequence; both now call SendService.admitOnce, keeping the charge-once
invariant in one place.

No behavior change; messaging tests pass (14/14).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* use explicit return statement

* Make comments mroe concise

* refactor: give each rate-limit config field a single responsibility

`isEnforcing` folded three fields into one "should I enforce" answer, so a
zeroed `messagesPerEpoch` or `epochPeriodSec` silently disabled an enabled
config. Split the responsibilities:

- `enabled` alone gates enforcement; `admit` reads it directly and
  `isEnforcing` is removed.
- a zero `messagesPerEpoch` now means what it says — admit nothing — which
  `admit` already yields (`0 >= 0` -> OverBudget), no special case.
- `RateLimitManager.new` returns a Result and rejects an enabled config with
  `epochPeriodSec == 0`, the only value that could crash the wall-clock
  fallback (`unixTime div epochPeriodSec`).

Callers thread the Result through; a zero-init `RateLimitConfig` stays valid
(disabled), so default construction is unaffected.

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

---------

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

logos_delivery/api/conf/messaging_conf.nim

@@ -4,9 +4,9 @@ import results, 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
 
 # `LogosDeliveryMode` and `EntryLayer` are defined at the leaf (`cli_args`) so
 # they can appear on `WakuNodeConf`; re-exported here via `kernel_conf`.
@@ -52,9 +52,12 @@ type MessagingClientConf* = object
   nodeKey* {.name: "nodekey".}: Opt[crypto.PrivateKey]
     ## P2P node private key (64-char hex): stable identity / peerId across restarts.
   rateLimit*: Opt[RateLimitConfig]
-    ## Per-epoch message rate limit enforced by the send service. `Opt` like
-    ## every other field so `merge` propagates a caller's override; unset falls
+    ## Per-epoch message rate limit enforced by the send service; unset falls
     ## back to `DefaultRateLimitConfig` (rate limiting disabled).
+    ##
+    ## Settable only programmatically: as a nested object with no `{.name.}`
+    ## pragma or `parseCmdArg`, it is not reachable from the JSON config or a
+    ## CLI flag.
 
 proc applyMode*(conf: var WakuNodeConf, mode: LogosDeliveryMode): ConfResult[void] =
   ## Sets the protocol flags implied by the mode.

logos_delivery/messaging/delivery_service/send_service/delivery_task.nim

@@ -26,6 +26,11 @@ type DeliveryTask* = ref object
   firstPropagatedTime*: Opt[Moment]
     ## Set once on the first successful propagation; never reset on re-publish.
     ## Anchors the store-validation time cap (see propagationAge).
+  firstAdmittedTime*: Opt[Moment]
+    ## Set when the task first passes rate-limit admission; `none` while parked
+    ## waiting for epoch budget. Guards re-admission on retry and anchors the
+    ## delivery-timeout reaper, so a task parked for budget is not aged out
+    ## before it can be sent.
   propagateEventEmitted*: bool
   errorDesc*: string
 
@@ -59,30 +64,47 @@ proc new*(
 
 func `==`*(r, l: DeliveryTask): bool =
   if r.isNil() == l.isNil():
-    r.isNil() or r.msgHash == l.msgHash
+    return r.isNil() or r.msgHash == l.msgHash
   else:
-    false
+    return false
 
 proc messageAge*(self: DeliveryTask): timer.Duration =
   let actual = getNanosecondTime(getTime().toUnixFloat())
   if self.msg.timestamp >= 0 and self.msg.timestamp < actual:
-    nanoseconds(actual - self.msg.timestamp)
+    return nanoseconds(actual - self.msg.timestamp)
   else:
-    ZeroDuration
+    return ZeroDuration
 
 proc deliveryAge*(self: DeliveryTask): timer.Duration =
   if self.state == DeliveryState.SuccessfullyPropagated:
-    timer.Moment.now() - self.deliveryTime
+    return timer.Moment.now() - self.deliveryTime
   else:
-    ZeroDuration
+    return ZeroDuration
 
 proc propagationAge*(self: DeliveryTask): timer.Duration =
   ## Time elapsed since the message was first successfully propagated.
   ## Stable across re-publishes; ZeroDuration until first propagation.
   if self.firstPropagatedTime.isSome():
-    timer.Moment.now() - self.firstPropagatedTime.get()
+    return timer.Moment.now() - self.firstPropagatedTime.get()
   else:
-    ZeroDuration
+    return ZeroDuration
+
+proc admissionAge*(self: DeliveryTask): timer.Duration =
+  ## Time since the task first passed admission; ZeroDuration while never
+  ## admitted (still parked waiting for epoch budget).
+  if self.firstAdmittedTime.isSome():
+    return timer.Moment.now() - self.firstAdmittedTime.get()
+  else:
+    return ZeroDuration
+
+proc isDeliveryTimedOut*(self: DeliveryTask, maxTime: timer.Duration): bool =
+  ## True when an admitted task has been trying to deliver longer than `maxTime`
+  ## without ever propagating. A task never admitted (parked for budget) is
+  ## exempt: the clock runs from admission time, so waiting for budget does not count
+  ## against it.
+  return
+    self.firstAdmittedTime.isSome() and self.firstPropagatedTime.isNone() and
+    self.admissionAge() > maxTime
 
 proc isEphemeral*(self: DeliveryTask): bool =
   return self.msg.ephemeral

logos_delivery/messaging/delivery_service/send_service/send_service.nim

@@ -48,8 +48,8 @@ type SendService* = ref object of RootObj
   serviceLoopHandle: Future[void] ## handle that allows to stop the async task
   sendProcessor: BaseSendProcessor
   rateLimitManager: RateLimitManager
-    ## Meters first transmissions against the per-epoch budget; re-publishes
-    ## of an already-propagated message resend the same bytes and are free.
+    ## Charges first transmissions against the per-epoch budget; re-publishes
+    ## are free.
 
   waku: Waku
   checkStoreForMessages: bool
@@ -85,7 +85,10 @@ proc new*(
     preferP2PReliability: bool,
     waku: Waku,
     rateLimitManager: RateLimitManager,
+    sendProcessor: BaseSendProcessor = nil,
 ): Result[T, string] =
+  ## `sendProcessor` overrides the relay/lightpush chain built from `waku`,
+  ## letting a caller drive the scheduler against a scripted delivery outcome.
   if not waku.hasRelay() and not waku.hasLightpush():
     return err(
       "Could not create SendService. wakuRelay or wakuLightpushClient should be set"
@@ -93,8 +96,12 @@ proc new*(
 
   let checkStoreForMessages = preferP2PReliability and waku.isStoreMounted()
 
-  let sendProcessorChain = setupSendProcessorChain(waku, waku.brokerCtx).valueOr:
-    return err("failed to setup SendProcessorChain: " & $error)
+  let sendProcessorChain =
+    if sendProcessor.isNil():
+      setupSendProcessorChain(waku, waku.brokerCtx).valueOr:
+        return err("failed to setup SendProcessorChain: " & $error)
+    else:
+      sendProcessor
 
   let sendService = SendService(
     brokerCtx: waku.brokerCtx,
@@ -197,15 +204,14 @@ proc reportTaskResult(self: SendService, task: DeliveryTask) =
     # rest of the states are intermediate and does not translate to event
     discard
 
-  # Only tasks that never propagated are reported as hard send failures here.
-  # Propagated-but-not-store-validated tasks are handled (warn + drop, no event)
-  # in evaluateAndCleanUp.
-  if task.firstPropagatedTime.isNone() and task.messageAge() > MaxTimeInCache:
+  # Hard-fail a task admitted but never propagated within MaxTimeInCache.
+  # Propagated-but-unvalidated tasks are dropped in evaluateAndCleanUp instead.
+  if task.isDeliveryTimedOut(MaxTimeInCache):
     error "Failed to send message",
       requestId = task.requestId,
       msgHash = task.msgHash.to0xHex(),
       error = "Message too old",
-      age = task.messageAge()
+      age = task.admissionAge()
     task.state = DeliveryState.FailedToDeliver
     MessageErrorEvent.emit(
       self.brokerCtx,
@@ -248,17 +254,27 @@ proc evaluateAndCleanUp(self: SendService) =
     )
   )
 
-proc trySendMessages(self: SendService) {.async.} =
+proc admitOnce(self: SendService, task: DeliveryTask): Future[bool] {.async.} =
+  ## Charges the task's first transmission against the epoch budget, at most
+  ## once per task (`firstAdmittedTime`); retries then resend for free. Returns
+  ## false when the task must stay parked for a later epoch.
+  if task.firstAdmittedTime.isSome():
+    return true
+
+  (await self.rateLimitManager.admit(task.msg.payload)).isOkOr:
+    debug "over rate-limit budget, task waits for the epoch to roll",
+      requestId = task.requestId, msgHash = task.msgHash.to0xHex()
+    return false
+  task.firstAdmittedTime = Opt.some(Moment.now())
+  return true
+
+proc trySendMessages*(self: SendService) {.async.} =
   let tasksToSend = self.taskCache.filterIt(it.state == DeliveryState.NextRoundRetry)
 
   for task in tasksToSend:
     # Todo, check if it has any perf gain to run them concurrent...
-    if task.firstPropagatedTime.isNone():
-      ## Never propagated: this transmission consumes a fresh epoch slot, so
-      ## it must be admitted. Tasks that stay over budget remain in
-      ## `NextRoundRetry` and are retried as the epoch rolls over.
-      (await self.rateLimitManager.admit(task.msg.payload)).isOkOr:
-        continue
+    if not (await self.admitOnce(task)):
+      continue
     await self.sendProcessor.process(task)
 
 proc serviceLoop(self: SendService) {.async.} =
@@ -288,8 +304,8 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} =
     error "SendService.send: failed to subscribe to content topic",
       contentTopic = task.msg.contentTopic, error = error
 
-  (await self.rateLimitManager.admit(task.msg.payload)).isOkOr:
-    info "SendService.send: over rate-limit budget, parking task",
+  if not (await self.admitOnce(task)):
+    info "SendService.send: parking task for a later round",
       requestId = task.requestId, msgHash = task.msgHash.to0xHex()
     task.state = DeliveryState.NextRoundRetry
     self.addTask(task)

logos_delivery/messaging/messaging_client.nim

@@ -6,6 +6,7 @@ import
   logos_delivery/api/conf/messaging_conf,
   logos_delivery/api/messaging_client_api,
   logos_delivery/waku/waku,
+  logos_delivery/waku/api/publish,
   logos_delivery/waku/factory/conf_builder/waku_conf_builder,
   logos_delivery/messaging/delivery_service/[recv_service, send_service],
   logos_delivery/messaging/rate_limit_manager/rate_limit_manager
@@ -19,15 +20,26 @@ type MessagingClient* = ref object
   recvService*: RecvService
   started*: bool
 
+proc rlnQuotaProvider(waku: Waku): QuotaProvider =
+  ## Sources the rate limit manager's epoch and limit from RLN. The closure
+  ## queries `waku` on each admission, so a node whose RLN mounts after
+  ## construction upgrades from the wall-clock fallback automatically.
+  return proc(): Opt[EpochQuota] {.gcsafe, raises: [].} =
+    let q = waku.currentRlnEpochQuota().valueOr:
+      return Opt.none(EpochQuota)
+    return
+      Opt.some(EpochQuota(epochIndex: q.epochIndex, userMessageLimit: q.messageLimit))
+
 proc new*(
     T: type MessagingClient, conf: MessagingClientConf, waku: Waku
 ): Result[T, string] =
   ## The messaging layer chains onto Waku: it drives the underlying Waku kernel
   ## for transport while exposing its own send/recv API.
   let reliability = conf.reliabilityEnabled.get(DefaultP2pReliability)
-  let sendService = ?SendService.new(
-    reliability, waku, RateLimitManager.new(conf.rateLimit.get(DefaultRateLimitConfig))
+  let rateLimitManager = ?RateLimitManager.new(
+    conf.rateLimit.get(DefaultRateLimitConfig), rlnQuotaProvider(waku)
   )
+  let sendService = ?SendService.new(reliability, waku, rateLimitManager)
   let recvService = RecvService.new(waku)
   return ok(
     T(

logos_delivery/messaging/rate_limit_manager/quota_source.nim

@@ -0,0 +1,22 @@
+## Epoch + user-message-limit source for the rate limit manager.
+##
+## Both values are read through one provider call, so a read cannot pair a
+## fresh epoch with a stale limit. The provider is a callback, keeping the
+## manager free of any `Waku` dependency.
+
+import std/times
+import results
+
+type
+  EpochQuota* = object
+    epochIndex*: uint64 ## Current epoch (`timestamp div` epoch length).
+    userMessageLimit*: uint64 ## Messages the epoch grants.
+
+  QuotaProvider* = proc(): Opt[EpochQuota] {.gcsafe, raises: [].}
+    ## `none` when RLN is not mounted — the signal to fall back to the
+    ## wall clock.
+
+proc wallClockEpochIndex*(epochPeriodSec: uint64): uint64 =
+  ## Absolute epoch (`unixTime div epochPeriodSec`), the same derivation RLN
+  ## uses, so independent nodes agree on the boundary.
+  return uint64(getTime().toUnix()) div epochPeriodSec

logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim

@@ -0,0 +1,25 @@
+## Configuration for the messaging rate limit manager.
+##
+## Kept separate from `rate_limit_manager` so `messaging_conf` can depend on
+## `RateLimitConfig` without pulling in the manager itself.
+
+type
+  RateLimitError* {.pure.} = enum
+    OverBudget
+
+  RateLimitConfig* = object
+    enabled*: bool
+    epochPeriodSec*: uint64
+      ## Epoch length in seconds. Shapes only the wall-clock fallback window;
+      ## ignored once the RLN quota source supplies the period.
+    messagesPerEpoch*: uint64
+      ## Local cap on messages admitted per epoch. When RLN is mounted the cap
+      ## is clamped to RLN's limit
+
+const
+  DefaultEpochPeriodSec* = 600'u64
+  DefaultMessagesPerEpoch* = 1'u64
+
+  DefaultRateLimitConfig* = RateLimitConfig(
+    epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch
+  ) ## Used when no rate-limit config is supplied; `enabled` defaults false.

logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim

@@ -1,58 +1,80 @@
 ## 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.
+## Rate-limits message transmissions against the per-epoch user message limit,
+## rejecting admission once the epoch's budget is spent. The epoch rolling
+## over refills the budget.
 ##
-## 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.
-##
-## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html
+## The epoch and limit come from a `QuotaProvider` when RLN is mounted;
+## otherwise a wall-clock window and the configured limit stand in. Parking and
+## retrying over-budget messages is the send service's job — this module only
+## answers whether one more transmission fits the current epoch.
 
-import std/times
 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
-    queue*: seq[seq[byte]]
-    currentEpochStart*: Time
-    sentInCurrentEpoch*: int
+type RateLimitManager* = ref object
+  config*: RateLimitConfig
+  quotaProvider: QuotaProvider
+    ## Nil or a `none` result selects the wall-clock fallback. Queried per
+    ## admission so a late RLN mount upgrades automatically.
+  currentEpochIndex*: uint64
+  sentInCurrentEpoch*: uint64
 
-const
-  DefaultEpochPeriodSec* = 600
-  DefaultMessagesPerEpoch* = 1
+proc new*(
+    T: type RateLimitManager,
+    config: RateLimitConfig,
+    quotaProvider: QuotaProvider = nil,
+): Result[T, string] =
+  ## Rejects an enabled config with a zero epoch period: the wall-clock
+  ## fallback derives the epoch as `unixTime div epochPeriodSec`.
+  if config.enabled and config.epochPeriodSec == 0:
+    return err("rate limit config: epochPeriodSec must be positive when enabled")
 
-  DefaultRateLimitConfig* = RateLimitConfig(
-    epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch
-  ) ## Used when no rate-limit config is supplied; `enabled` defaults false.
+  return ok(
+    T(
+      config: config,
+      quotaProvider: quotaProvider,
+      currentEpochIndex: 0,
+      sentInCurrentEpoch: 0,
+    )
+  )
 
-proc new*(T: type RateLimitManager, config: RateLimitConfig): T =
-  return
-    T(config: config, queue: @[], currentEpochStart: getTime(), sentInCurrentEpoch: 0)
+proc currentQuota(self: RateLimitManager): Opt[EpochQuota] =
+  if self.quotaProvider.isNil():
+    return Opt.none(EpochQuota)
+  return self.quotaProvider()
 
 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`.
+  ## Charges one message against the current epoch's limit, rolling the window
+  ## first when the epoch has advanced. A disabled config admits everything.
+  if not self.config.enabled:
+    return ok()
+
+  let quota = self.currentQuota()
+
+  let epochIndex =
+    if quota.isSome():
+      quota.get().epochIndex
+    else:
+      wallClockEpochIndex(self.config.epochPeriodSec)
+
+  # RLN can only tighten the configured cap, never widen it: exceeding RLN's
+  # limit would fail later at proof generation.
+  var limit = self.config.messagesPerEpoch
+  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)
+
+  self.sentInCurrentEpoch.inc()
   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
-
-proc resetEpoch*(self: RateLimitManager) =
-  self.currentEpochStart = getTime()
-  self.sentInCurrentEpoch = 0

logos_delivery/waku/api/publish.nim

@@ -42,6 +42,17 @@ proc relayPushHandler*(self: Waku): PushMessageHandler =
   ## in (legacy)lightpushPublish; this handler only validates and republishes.
   return getRelayPushHandler(self.node.wakuRelay)
 
+proc currentRlnEpochQuota*(self: Waku): Opt[tuple[epochIndex, messageLimit: uint64]] =
+  ## RLN's current epoch index and user message limit, read together so the
+  ## pair cannot straddle an epoch boundary.
+  if self.node.rln.isNil():
+    return Opt.none(tuple[epochIndex, messageLimit: uint64])
+
+  let limit = self.node.rln.groupManager.userMessageLimit.valueOr:
+    return Opt.none(tuple[epochIndex, messageLimit: uint64])
+
+  return Opt.some((fromEpoch(self.node.rln.getCurrentEpoch()), uint64(limit)))
+
 proc lightpushPeerAvailable*(self: Waku, shard: PubsubTopic): bool =
   ## True if a lightpush service peer is available for `shard`.
   return self.node.peerManager.selectPeer(WakuLightPushCodec, Opt.some(shard)).isSome()

tests/messaging/test_all.nim

@@ -1,3 +1,4 @@
 {.used.}
 
-import ./test_rate_limit_manager
+import
+  ./test_rate_limit_manager, ./test_delivery_task_reaping, ./test_send_service_scheduler

tests/messaging/test_delivery_task_reaping.nim

@@ -0,0 +1,36 @@
+{.used.}
+
+import results, chronos, testutils/unittests
+
+import logos_delivery/messaging/delivery_service/send_service/delivery_task
+
+const MaxTime = chronos.minutes(1)
+
+proc taskWith(admitted, propagated: Opt[Moment]): DeliveryTask =
+  ## Builds a DeliveryTask directly (bypassing `new`, which needs a broker).
+  return DeliveryTask(firstAdmittedTime: admitted, firstPropagatedTime: propagated)
+
+suite "DeliveryTask - delivery-timeout reaping":
+  test "a task parked for budget (never admitted) is exempt, however old":
+    ## Budget-parked tasks must survive to the epoch roll, not be aged out.
+    let task = taskWith(Opt.none(Moment), Opt.none(Moment))
+    check not task.isDeliveryTimedOut(MaxTime)
+
+  test "an admitted, never-propagated task past the window times out":
+    let task = taskWith(Opt.some(Moment.now() - chronos.minutes(2)), Opt.none(Moment))
+    check task.isDeliveryTimedOut(MaxTime)
+
+  test "an admitted task still within the window does not time out":
+    let task = taskWith(Opt.some(Moment.now()), Opt.none(Moment))
+    check not task.isDeliveryTimedOut(MaxTime)
+
+  test "a propagated task is never timed out here (store validation owns it)":
+    let task =
+      taskWith(Opt.some(Moment.now() - chronos.minutes(2)), Opt.some(Moment.now()))
+    check not task.isDeliveryTimedOut(MaxTime)
+
+  test "the timeout clock runs from admission, not message creation":
+    ## A just-admitted task has a fresh clock, even after a long budget wait.
+    let task = taskWith(Opt.some(Moment.now()), Opt.none(Moment))
+    check task.admissionAge() < MaxTime
+    check not task.isDeliveryTimedOut(MaxTime)

tests/messaging/test_rate_limit_manager.nim

@@ -1,26 +1,98 @@
 {.used.}
 
-import chronos, testutils/unittests, stew/byteutils
+import results, 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, so limit-boundary tests don't touch
+  ## the wall clock.
+  return proc(): Opt[EpochQuota] {.gcsafe, raises: [].} =
+    return
+      Opt.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)
-    )
+    let rl = RateLimitManager
+      .new(RateLimitConfig(enabled: false, epochPeriodSec: 600, messagesPerEpoch: 1))
+      .expect("RateLimitManager.new")
     for _ in 0 ..< 10:
-      let res = await rl.admit("payload".toBytes())
-      check res.isOk()
+      check (await rl.admit("payload".toBytes())).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.
-    let rl = RateLimitManager.new(
-      RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1)
-    )
+  asyncTest "a zero cap admits nothing":
+    let rl = RateLimitManager
+      .new(RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 0))
+      .expect("RateLimitManager.new")
+    let res = await rl.admit("a".toBytes())
+    check:
+      res.isErr()
+      res.error == RateLimitError.OverBudget
+
+  test "construction validates the epoch period only when enabled":
+    check:
+      RateLimitManager
+        .new(RateLimitConfig(enabled: true, epochPeriodSec: 0, messagesPerEpoch: 1))
+        .isErr()
+      RateLimitManager.new(RateLimitConfig()).isOk()
+
+  asyncTest "admits up to the 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),
+        fixedQuota(epochIndex = 42, userMessageLimit = 100),
+      )
+      .expect("RateLimitManager.new")
+    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 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: 600, messagesPerEpoch: 1),
+        proc(): Opt[EpochQuota] {.gcsafe, raises: [].} =
+          Opt.some(EpochQuota(epochIndex: epoch, userMessageLimit: 100)),
+      )
+      .expect("RateLimitManager.new")
     check (await rl.admit("first".toBytes())).isOk()
-    check (await rl.admit("second".toBytes())).isOk()
+    check (await rl.admit("second".toBytes())).isErr()
+    epoch = 2
+    check (await rl.admit("third".toBytes())).isOk()
+    check (await rl.admit("fourth".toBytes())).isErr()
+
+  asyncTest "RLN user message limit clamps a looser configured cap":
+    ## config cap 5, RLN grants 2 — the lower RLN limit wins.
+    let rl = RateLimitManager
+      .new(
+        RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 5),
+        fixedQuota(epochIndex = 7, userMessageLimit = 2),
+      )
+      .expect("RateLimitManager.new")
+    check (await rl.admit("a".toBytes())).isOk()
+    check (await rl.admit("b".toBytes())).isOk()
+    check (await rl.admit("c".toBytes())).isErr()
+
+  asyncTest "config cap can tighten below the RLN limit":
+    ## config cap 1, RLN grants 100 — the config cap wins.
+    let rl = RateLimitManager
+      .new(
+        RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1),
+        fixedQuota(epochIndex = 7, userMessageLimit = 100),
+      )
+      .expect("RateLimitManager.new")
+    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, epochPeriodSec: 600, messagesPerEpoch: 1))
+      .expect("RateLimitManager.new")
+    check (await rl.admit("first".toBytes())).isOk()
+    check (await rl.admit("second".toBytes())).isErr()

tests/messaging/test_send_service_scheduler.nim

@@ -0,0 +1,150 @@
+{.used.}
+
+import std/net
+import chronos, chronicles, testutils/unittests, results, stew/byteutils
+
+import
+  logos_delivery/waku/waku,
+  logos_delivery/waku/waku_core,
+  logos_delivery/api/types,
+  logos_delivery/api/conf/messaging_conf,
+  logos_delivery/waku/factory/waku_conf,
+  logos_delivery/messaging/rate_limit_manager/rate_limit_manager,
+  logos_delivery/messaging/delivery_service/send_service/
+    [send_service, send_processor, delivery_task]
+import ../testlib/testasync
+
+## Scheduler-level coverage for the send service's rate-limit seam: a task is
+## charged exactly once however many rounds it takes, and an over-budget task
+## is parked then released when the epoch rolls. A fake processor scripts the
+## delivery outcome so the loop runs without network or sleeps.
+
+type FakeSendProcessor = ref object of BaseSendProcessor
+  calls: int
+  script: seq[DeliveryState]
+    ## State to stamp on the task per invocation; the last entry repeats.
+
+method process(self: FakeSendProcessor, task: DeliveryTask): Future[void] {.async.} =
+  let outcome = self.script[min(self.calls, self.script.high)]
+  inc self.calls
+  task.state = outcome
+  if outcome == DeliveryState.SuccessfullyPropagated and
+      task.firstPropagatedTime.isNone():
+    task.firstPropagatedTime = Opt.some(Moment.now())
+
+proc testConf(): WakuConf =
+  var conf = MessagingClientConf()
+    .toWakuNodeConf(messaging_conf.LogosDeliveryMode.Core).valueOr:
+      raiseAssert error
+  conf.listenAddress = parseIpAddress("0.0.0.0")
+  conf.tcpPort = Port(0)
+  conf.discv5UdpPort = Port(0)
+  conf.clusterId = Opt.some(3'u16)
+  conf.numShardsInNetwork = 1
+  conf.rest = false
+  return conf.toWakuConf().valueOr:
+    raiseAssert error
+
+proc fixedEpochQuota(epoch: ptr uint64, userMessageLimit: uint64): QuotaProvider =
+  ## Quota pinned to whatever `epoch` holds, so a test rolls the epoch by
+  ## writing through the pointer.
+  return proc(): Opt[EpochQuota] {.gcsafe, raises: [].} =
+    return Opt.some(EpochQuota(epochIndex: epoch[], userMessageLimit: userMessageLimit))
+
+suite "SendService - rate-limit scheduling":
+  var waku {.threadvar.}: Waku
+
+  asyncSetup:
+    waku = (await Waku.new(testConf())).expect("Waku.new")
+
+  asyncTeardown:
+    ## The node is never started, so stop is best-effort cleanup.
+    discard await waku.stop()
+
+  proc buildTask(id, payload: string): DeliveryTask =
+    ## Built directly rather than via `DeliveryTask.new`, which needs a broker
+    ## provider only registered once the node starts.
+    let msg = WakuMessage(
+      contentTopic: "/test/1/scheduler/proto",
+      payload: payload.toBytes(),
+      timestamp: 1_700_000_000_000_000_000,
+    )
+    let pubsubTopic = PubsubTopic("/waku/2/rs/3/0")
+    return DeliveryTask(
+      requestId: RequestId(id),
+      pubsubTopic: pubsubTopic,
+      msg: msg,
+      msgHash: computeMessageHash(pubsubTopic, msg),
+      state: DeliveryState.Entry,
+    )
+
+  asyncTest "a task is charged once even when delivery takes several rounds":
+    ## First round fails to propagate, second succeeds. The retry must not draw a
+    ## second slot: `firstAdmittedTime` guards re-admission.
+    var epoch = 5'u64
+    let manager = RateLimitManager
+      .new(
+        RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 3),
+        fixedEpochQuota(addr epoch, userMessageLimit = 100),
+      )
+      .expect("RateLimitManager.new")
+    let processor = FakeSendProcessor(
+      script: @[DeliveryState.NextRoundRetry, DeliveryState.SuccessfullyPropagated]
+    )
+    let service =
+      SendService.new(false, waku, manager, processor).expect("SendService.new")
+
+    let task = buildTask("charge-once", "hi")
+    await service.send(task)
+    check:
+      manager.sentInCurrentEpoch == 1'u64
+      task.firstAdmittedTime.isSome()
+      task.state == DeliveryState.NextRoundRetry
+
+    await service.trySendMessages()
+    check:
+      manager.sentInCurrentEpoch == 1'u64 # not re-charged on retry
+      processor.calls == 2
+      task.state == DeliveryState.SuccessfullyPropagated
+
+  asyncTest "an over-budget task is parked, then released when the epoch rolls":
+    ## Budget of one per epoch. The second send is parked until the epoch rolls,
+    ## then admitted and delivered.
+    var epoch = 1'u64
+    let manager = RateLimitManager
+      .new(
+        RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1),
+        fixedEpochQuota(addr epoch, userMessageLimit = 100),
+      )
+      .expect("RateLimitManager.new")
+    let processor = FakeSendProcessor(script: @[DeliveryState.SuccessfullyPropagated])
+    let service =
+      SendService.new(false, waku, manager, processor).expect("SendService.new")
+
+    let first = buildTask("in-budget", "one")
+    await service.send(first)
+    check:
+      first.state == DeliveryState.SuccessfullyPropagated
+      manager.sentInCurrentEpoch == 1'u64
+
+    let second = buildTask("over-budget", "two")
+    await service.send(second)
+    check:
+      second.state == DeliveryState.NextRoundRetry # parked
+      second.firstAdmittedTime.isNone() # never admitted
+    let callsWhenParked = processor.calls
+
+    # Same epoch: still over budget, so the parked task is not handed to the
+    # processor.
+    await service.trySendMessages()
+    check:
+      second.state == DeliveryState.NextRoundRetry
+      second.firstAdmittedTime.isNone()
+      processor.calls == callsWhenParked
+
+    # Epoch rolls: budget refills, the parked task is admitted and delivered.
+    epoch = 2'u64
+    await service.trySendMessages()
+    check:
+      second.firstAdmittedTime.isSome()
+      second.state == DeliveryState.SuccessfullyPropagated