feat: meter SendService transmissions through RateLimitManager

Wires the relocated RateLimitManager into the messaging layer at the
transmission stage rather than the API entry point:

- MessagingClientConf gains a rateLimit: RateLimitConfig field
  (defaulting to DefaultEpochPeriodSec / DefaultMessagesPerEpoch), and
  MessagingClient.new hands the constructed manager to SendService.
- SendService consults admit() before the first transmission of a
  task, both in send() and in the retry loop. Re-publishes of an
  already-propagated message (firstPropagatedTime set) skip admission:
  they resend the same bytes, which reuse the same RLN proof/nullifier
  and consume no fresh epoch slot.
- An over-budget task parks in the task cache as NextRoundRetry; the
  service loop re-admits it as the epoch budget frees up. The skeleton
  admit() is a pass-through, so behaviour is unchanged today.

Gating transmissions instead of MessagingClient.send keeps SDS repair
rebroadcasts free of API-entry rejection (SDS decides that a repair is
needed; the transmission scheduler decides when it fits the budget)
while every wire transmission still draws from one node-wide budget --
which network-side RLN enforcement applies to repairs regardless of
any local bypass. It is also where RLN proof attachment must happen,
since proofs bind to the epoch current at transmission time.

Adds tests/messaging/test_rate_limit_manager.nim covering the current
disabled + enabled pass-through behaviour, wired into
all_tests_waku.nim.

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

Author: stubbsta

logos_delivery/api/conf/messaging_conf.nim

@@ -6,8 +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
 
-export kernel_conf
+export kernel_conf, rate_limit_manager
 
 type LogosDeliveryMode* {.pure.} = enum
   Edge # client-only node
@@ -54,6 +55,9 @@ type MessagingClientConf* = object
     ## Process log format (TEXT or JSON); applied by the kernel on node creation.
   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.
 
 proc applyMode*(conf: var WakuNodeConf, mode: LogosDeliveryMode): ConfResult[void] =
   ## Sets the protocol flags implied by the mode.

logos_delivery/messaging/delivery_service/send_service/send_service.nim

@@ -9,7 +9,8 @@ import
   ./[send_processor, relay_processor, lightpush_processor, delivery_task],
   logos_delivery/waku/[waku_core, waku_store/common],
   logos_delivery/waku/waku,
-  logos_delivery/waku/api/[store, subscriptions, publish]
+  logos_delivery/waku/api/[store, subscriptions, publish],
+  logos_delivery/messaging/rate_limit_manager/rate_limit_manager
 import logos_delivery/api/events/messaging_client_events
 
 logScope:
@@ -47,6 +48,9 @@ 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.
 
   waku: Waku
   checkStoreForMessages: bool
@@ -78,7 +82,10 @@ proc setupSendProcessorChain(
   return ok(processors[0])
 
 proc new*(
-    T: typedesc[SendService], preferP2PReliability: bool, waku: Waku
+    T: typedesc[SendService],
+    preferP2PReliability: bool,
+    waku: Waku,
+    rateLimitManager: RateLimitManager,
 ): Result[T, string] =
   if not waku.hasRelay() and not waku.hasLightpush():
     return err(
@@ -95,6 +102,7 @@ proc new*(
     taskCache: newSeq[DeliveryTask](),
     serviceLoopHandle: nil,
     sendProcessor: sendProcessorChain,
+    rateLimitManager: rateLimitManager,
     waku: waku,
     checkStoreForMessages: checkStoreForMessages,
     lastStoreCheckTime: Moment.now(),
@@ -246,6 +254,12 @@ proc trySendMessages(self: SendService) {.async.} =
 
   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
     await self.sendProcessor.process(task)
 
 proc serviceLoop(self: SendService) {.async.} =
@@ -275,6 +289,13 @@ 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",
+      requestId = task.requestId, msgHash = task.msgHash.to0xHex()
+    task.state = DeliveryState.NextRoundRetry
+    self.addTask(task)
+    return
+
   await self.sendProcessor.process(task)
   reportTaskResult(self, task)
   if task.state != DeliveryState.FailedToDeliver:

logos_delivery/messaging/messaging_client.nim

@@ -8,7 +8,8 @@ import
   logos_delivery/api/messaging_client_api,
   logos_delivery/waku/waku,
   logos_delivery/waku/factory/conf_builder/waku_conf_builder,
-  logos_delivery/messaging/delivery_service/[recv_service, send_service]
+  logos_delivery/messaging/delivery_service/[recv_service, send_service],
+  logos_delivery/messaging/rate_limit_manager/rate_limit_manager
 
 export messaging_client_api, messaging_conf
 
@@ -25,7 +26,8 @@ proc new*(
   ## 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)
+  let sendService =
+    ?SendService.new(reliability, waku, RateLimitManager.new(conf.rateLimit))
   let recvService = RecvService.new(waku)
   return ok(
     T(

tests/all_tests_waku.nim

@@ -88,5 +88,8 @@ import ./tools/test_all
 # Persistency library tests
 import ./persistency/test_all
 
+# Messaging API tests
+import ./messaging/test_all
+
 # Reliable Channel API tests
 import ./channels/test_all

tests/messaging/test_all.nim

@@ -0,0 +1,3 @@
+{.used.}
+
+import ./test_rate_limit_manager

tests/messaging/test_rate_limit_manager.nim

@@ -0,0 +1,26 @@
+{.used.}
+
+import chronos, testutils/unittests, stew/byteutils
+
+import logos_delivery/messaging/rate_limit_manager/rate_limit_manager
+
+suite "RateLimitManager - admission":
+  asyncTest "admit is a pass-through when disabled":
+    let rl = RateLimitManager.new(
+      RateLimitConfig(enabled: false, epochPeriodSec: 600, messagesPerEpoch: 1)
+    )
+    for _ in 0 ..< 10:
+      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.
+    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()