Merge 0674efff41cbf41a04be7832833639df6fcb607d into 9bf7e57f922ad8b8163f5c3603d2cd036e6c44f0

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/delivery_service/send_service/lightpush_processor.nim

@@ -36,6 +36,17 @@ method sendImpl*(
     await self.waku.lightpushPublishToAny(task.pubsubTopic, task.msg)
   ).valueOr:
     error "LightpushSendProcessor.sendImpl failed", error = error.desc.get($error.code)
+
+    if error.isRlnRejection():
+      ## The proof was refused, so it must not be sent again: drop it and let
+      ## the refreshed merkle path produce a new one on the next round.
+      ## Re-admission gates the regeneration, so a task cannot spin through the
+      ## epoch budget by retrying.
+      self.waku.onRlnProofRejected()
+      task.msg.proof = @[]
+      task.state = DeliveryState.NextRoundRetry
+      return
+
     case error.code
     of LightPushErrorCode.NO_PEERS_TO_RELAY, LightPushErrorCode.TOO_MANY_REQUESTS,
         LightPushErrorCode.OUT_OF_RLN_PROOF, LightPushErrorCode.SERVICE_NOT_AVAILABLE,

logos_delivery/messaging/delivery_service/send_service/relay_processor.nim

@@ -3,6 +3,7 @@ import std/options
 import chronos, chronicles
 import brokers/broker_context
 import logos_delivery/waku/[waku_core], logos_delivery/waku/waku_lightpush/[common, rpc]
+import logos_delivery/waku/waku, logos_delivery/waku/api/publish
 import logos_delivery/waku/requests/health_requests
 import logos_delivery/api/types
 import ./[delivery_task, send_processor]
@@ -11,6 +12,7 @@ logScope:
   topics = "send service relay processor"
 
 type RelaySendProcessor* = ref object of BaseSendProcessor
+  waku: Waku
   publishProc: PushMessageHandler
   fallbackStateToSet: DeliveryState
 
@@ -18,6 +20,7 @@ proc new*(
     T: typedesc[RelaySendProcessor],
     lightpushAvailable: bool,
     publishProc: PushMessageHandler,
+    waku: Waku,
     brokerCtx: BrokerContext,
 ): RelaySendProcessor =
   let fallbackStateToSet =
@@ -27,6 +30,7 @@ proc new*(
       DeliveryState.FailedToDeliver
 
   return RelaySendProcessor(
+    waku: waku,
     publishProc: publishProc,
     fallbackStateToSet: fallbackStateToSet,
     brokerCtx: brokerCtx,
@@ -62,6 +66,17 @@ method sendImpl*(self: RelaySendProcessor, task: DeliveryTask) {.async.} =
     let errorMessage = error.desc.get($error.code)
     error "Failed to publish message with relay",
       request = task.requestId, msgHash = task.msgHash.to0xHex(), error = errorMessage
+
+    if error.isRlnRejection():
+      ## The relay validator refused the proof. Dropping it and retrying is not
+      ## the same as failing: the message is valid, its proof went stale against
+      ## a moved merkle root. Clearing it makes the next round regenerate one
+      ## against the refreshed path.
+      self.waku.onRlnProofRejected()
+      task.msg.proof = @[]
+      task.state = DeliveryState.NextRoundRetry
+      return
+
     if error.code != LightPushErrorCode.NO_PEERS_TO_RELAY:
       task.state = DeliveryState.FailedToDeliver
       task.errorDesc = errorMessage

logos_delivery/messaging/delivery_service/send_service/send_service.nim

@@ -69,7 +69,9 @@ proc setupSendProcessorChain(
 
   if isRelayAvail:
     let publishProc = waku.relayPushHandler()
-    processors.add(RelaySendProcessor.new(isLightPushAvail, publishProc, brokerCtx))
+    processors.add(
+      RelaySendProcessor.new(isLightPushAvail, publishProc, waku, brokerCtx)
+    )
   if isLightPushAvail:
     processors.add(LightpushSendProcessor.new(waku, brokerCtx))
 
@@ -260,6 +262,15 @@ proc trySendMessages(self: SendService) {.async.} =
       ## `NextRoundRetry` and are retried as the epoch rolls over.
       (await self.rateLimitManager.admit(task.msg.payload)).isOkOr:
         continue
+
+      ## Strictly after admission, so a rejected message never draws a nonce.
+      ## A no-op when RLN is not mounted, or when a prior round already
+      ## attached a proof.
+      task.msg = (await self.waku.attachRlnProof(task.msg)).valueOr:
+        error "SendService: failed to attach RLN proof, retrying next round",
+          requestId = task.requestId, error = error
+        continue
+
     await self.sendProcessor.process(task)
 
 proc serviceLoop(self: SendService) {.async.} =
@@ -296,6 +307,15 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} =
     self.addTask(task)
     return
 
+  ## Strictly after admission, so a rejected message never draws a nonce.
+  ## A no-op when RLN is not mounted.
+  task.msg = (await self.waku.attachRlnProof(task.msg)).valueOr:
+    error "SendService.send: failed to attach RLN proof, parking task",
+      requestId = task.requestId, error = error
+    task.state = DeliveryState.NextRoundRetry
+    self.addTask(task)
+    return
+
   await self.sendProcessor.process(task)
   reportTaskResult(self, task)
   if task.state != DeliveryState.FailedToDeliver:

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,54 +1,81 @@
 ## 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
+## 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.
 ##
-## 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.
+## 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
-    queue*: seq[seq[byte]]
-    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, queue: @[], currentEpochStart: getTime(), 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: []).} =
-  ## 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 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()
+
+  let quota = self.currentQuota()
+
+  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
   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

@@ -8,7 +8,7 @@
 import logos_delivery/waku/compat/option_valueor
 {.push raises: [].}
 
-import std/options
+import std/[options, times, strutils]
 import results, chronos
 
 import logos_delivery/waku/waku
@@ -40,10 +40,59 @@ proc hasLightpush*(self: Waku): bool =
 
 proc relayPushHandler*(self: Waku): PushMessageHandler =
   ## Builds the relay publish handler used by the send pipeline. Caller
-  ## ensures relay is mounted. RLN proof generation is handled client-side
-  ## in (legacy)lightpushPublish; this handler only validates and republishes.
+  ## ensures relay is mounted. The handler validates and republishes; the
+  ## proof is attached by the messaging layer via `attachRlnProof`.
   return getRelayPushHandler(self.node.wakuRelay)
 
+proc attachRlnProof*(
+    self: Waku, message: WakuMessage
+): Future[Result[WakuMessage, string]] {.async.} =
+  ## Returns `message` carrying an RLN proof. A message that already has one is
+  ## returned untouched, so retrying a task neither redraws a nonce nor changes
+  ## the bytes. Without RLN mounted the message passes through unproven.
+  ##
+  ## Uses the root-refreshing generator: a message can wait in the send
+  ## service's task cache while the group root moves on chain, so the proof is
+  ## validated against the acceptable-root window and regenerated once against a
+  ## refetched merkle path if it went stale.
+  if self.node.rln.isNil() or message.proof.len > 0:
+    return ok(message)
+
+  var msgWithProof = message
+  msgWithProof.proof = (
+    await self.node.rln.generateRLNProofWithRootRefresh(
+      message.toRLNSignal(), float64(getTime().toUnix())
+    )
+  ).valueOr:
+    return err("failed to attach RLN proof: " & error)
+
+  return ok(msgWithProof)
+
+func isRlnRejection*(error: ErrorStatus): bool =
+  ## True when a publish failure means "the RLN proof was not accepted", so the
+  ## message is worth retrying with a freshly generated proof rather than being
+  ## failed outright.
+  ##
+  ## OUT_OF_RLN_PROOF is always RLN. INVALID_MESSAGE also covers non-RLN
+  ## rejections (an oversized message, say), so it additionally has to carry the
+  ## validator's error marker — this is the same gate the kernel lightpush path
+  ## applies before scheduling a refresh.
+  return
+    error.code == LightPushErrorCode.OUT_OF_RLN_PROOF or (
+      error.code == LightPushErrorCode.INVALID_MESSAGE and
+      error.desc.get("").contains(RlnValidatorErrorMsg)
+    )
+
+proc onRlnProofRejected*(self: Waku) =
+  ## Called when a publish was rejected as RLN-invalid. Starts refetching the
+  ## merkle path in the background, so the next proof generated for the message
+  ## is built against a fresh one. Non-blocking: the send service's own loop is
+  ## what retries, and it must not stall waiting on an RPC round trip.
+  if self.node.rln.isNil():
+    return
+
+  self.node.rln.groupManager.scheduleMerkleProofRefresh()
+
 proc lightpushPeerAvailable*(self: Waku, shard: PubsubTopic): bool =
   ## True if a lightpush service peer is available for `shard`.
   return self.node.peerManager.selectPeer(WakuLightPushCodec, some(shard)).isSome()

logos_delivery/waku/node/waku_node/lightpush.nim

@@ -105,30 +105,6 @@ proc resolveLegacyPubsubTopic(
       return err("Autosharding error: " & error)
   return ok($shard)
 
-proc runRlnRefreshRetry(
-    node: WakuNode,
-    rln: Option[Rln],
-    msgWithProof: WakuMessage,
-    pubsubForPublish: PubsubTopic,
-    peer: RemotePeerInfo,
-    fallback: legacy_lightpush_protocol.WakuLightPushResult[string],
-): Future[legacy_lightpush_protocol.WakuLightPushResult[string]] {.async, gcsafe.} =
-  ## Refreshes the RLN merkle proof path and retries the publish once. Only the
-  ## refresh is bounded by RlnMerkleProofRefreshTimeout (returning `fallback` on
-  ## timeout); the retried publish runs unbounded, matching the first attempt.
-  info "legacy lightpush send rejected as RLN-invalid; " &
-    "refreshing merkle proof and retrying once"
-  rln.get().groupManager.invalidateMerkleProofCache()
-
-  let refreshFut = attachRLNProof(rln.get(), msgWithProof)
-  if not (await refreshFut.withTimeout(RlnMerkleProofRefreshTimeout)):
-    warn "legacy lightpush RLN proof refresh timed out; returning original error"
-    return fallback
-  let retryMsg = refreshFut.read().valueOr:
-    return err("failed call attachRLNProof from lightpush retry: " & error)
-
-  return await internalLegacyLightpushPublish(node, pubsubForPublish, retryMsg, peer)
-
 proc legacyLightpushPublish*(
     node: WakuNode,
     pubsubTopic: Option[PubsubTopic],
@@ -161,18 +137,20 @@ proc legacyLightpushPublish*(
     ).valueOr:
       return err(error)
 
-    let firstResult =
+    let publishResult =
       await internalLegacyLightpushPublish(node, pubsubForPublish, msgWithProof, peer)
 
     # Legacy has no status codes, so string-match the RLN error to detect a
-    # stale merkle proof path, then refresh and retry once.
-    if firstResult.isOk() or rln.isNone() or
-        not firstResult.error.contains(RlnValidatorErrorMsg):
-      return firstResult
+    # stale merkle proof path. Schedule the refresh and hand the error back:
+    # retrying is the caller's decision, the same way the non-legacy path
+    # behaves. A retry regenerates the proof against the refreshed cache.
+    if publishResult.isOk() or rln.isNone() or
+        not publishResult.error.contains(RlnValidatorErrorMsg):
+      return publishResult
 
-    return await runRlnRefreshRetry(
-      node, rln, msgWithProof, pubsubForPublish, peer, firstResult
-    )
+    info "legacy lightpush send rejected as RLN-invalid; scheduling merkle proof refresh"
+    rln.get().groupManager.scheduleMerkleProofRefresh()
+    return err(RlnProofRefreshScheduledMsg & ": " & publishResult.error)
   except CatchableError:
     return err(getCurrentExceptionMsg())
 

logos_delivery/waku/rln/constants.nim

@@ -25,11 +25,6 @@ const RlnValidatorErrorMsg* = "RLN validation failed"
 const RlnProofRefreshScheduledMsg* =
   "stale RLN proof suspected; refresh scheduled, retry the publish"
 
-# Bounds the legacy lightpush merkle proof refresh (eth_call refetch + proof
-# regen) so a hanging RPC cannot stall the caller. The retried publish is not
-# bounded.
-const RlnMerkleProofRefreshTimeout* = 5.seconds
-
 # inputs of the membership contract constructor
 # TODO may be able to make these constants private and put them inside the waku_rln_utils
 const

logos_delivery/waku/rln/proof.nim

@@ -54,23 +54,41 @@ proc toRLNSignal*(wakumessage: WakuMessage): seq[byte] =
     output = concat(wakumessage.payload, contentTopicBytes, @(timestampBytes))
   return output
 
-proc generateRLNProof*(
-    rln: Rln, input: seq[byte], senderEpochTime: float64
+proc generateRLNProofWithNonce(
+    rln: Rln, input: seq[byte], senderEpochTime: float64, nonce: Nonce
 ): Future[Result[seq[byte], string]] {.async.} =
+  ## Generates a proof against an already drawn `nonce`. Regenerating for an
+  ## unchanged (input, epoch, nonce) is safe: the revealed share is a function
+  ## of those three, so a regenerated proof reveals the same share and cannot
+  ## read as double-signalling.
   let epoch = rln.calcEpoch(senderEpochTime)
-  let nonce = rln.nonceManager.getNonce().valueOr:
-    return err("could not get new message id to generate an rln proof: " & $error)
   let proof = (await rln.groupManager.generateProof(input, epoch, nonce)).valueOr:
     return err("could not generate rln-v2 proof: " & $error)
   return ok(proof.encode().buffer)
 
+proc generateRLNProof*(
+    rln: Rln, input: seq[byte], senderEpochTime: float64
+): Future[Result[seq[byte], string]] {.async.} =
+  let nonce = rln.nonceManager.getNonce().valueOr:
+    return err("could not get new message id to generate an rln proof: " & $error)
+  return await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce)
+
 proc generateRLNProofWithRootRefresh*(
     rln: Rln, input: seq[byte], senderEpochTime: float64
 ): Future[Result[seq[byte], string]] {.async.} =
   ## Generates an RLN proof and checks its merkle root against the
   ## acceptable-root window. If the root is stale, invalidates the cache and
   ## regenerates once against a refetched path. Returns the proof bytes.
-  let proofBytes = (await rln.generateRLNProof(input, senderEpochTime)).valueOr:
+  ##
+  ## The regeneration reuses the nonce drawn for the first attempt: only the
+  ## merkle path differs between the two, so drawing again would spend two
+  ## message ids from the epoch budget on a message that is sent once. That
+  ## would drift the budget the rate limit manager accounts for away from the
+  ## one the nonce manager enforces.
+  let nonce = rln.nonceManager.getNonce().valueOr:
+    return err("could not get new message id to generate an rln proof: " & $error)
+
+  let proofBytes = (await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce)).valueOr:
     return err("failed to generate RLN proof: " & $error)
 
   let rlnProof = RateLimitProof.init(proofBytes).valueOr:
@@ -81,7 +99,7 @@ proc generateRLNProofWithRootRefresh*(
 
   info "RLN: stale merkle root detected; refreshing merkle path and regenerating proof"
   rln.groupManager.invalidateMerkleProofCache()
-  return await rln.generateRLNProof(input, senderEpochTime)
+  return await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce)
 
 proc attachRLNProof*(
     r: Rln, message: WakuMessage

tests/messaging/test_all.nim

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

tests/messaging/test_rate_limit_manager.nim

@@ -1,26 +1,87 @@
 {.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 "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 "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: 1)
+      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()
+    let res = await rl.admit("over".toBytes())
+    check:
+      res.isErr()
+      res.error == RateLimitError.OverBudget
+
+  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, 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())).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 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, 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()

tests/messaging/test_rln_proof_attach.nim

@@ -0,0 +1,102 @@
+{.used.}
+
+import std/[options, net, osproc]
+import chronos, testutils/unittests, results, stew/byteutils
+import
+  logos_delivery/waku/[waku, waku_core, rln],
+  logos_delivery/waku/node/waku_node,
+  logos_delivery/waku/node/waku_node/relay,
+  logos_delivery/waku/api/publish,
+  logos_delivery/api/conf/messaging_conf,
+  logos_delivery/waku/factory/waku_conf
+import
+  ../testlib/testasync,
+  ../waku_rln_relay/utils_onchain,
+  ../waku_rln_relay/rln/waku_rln_relay_utils
+
+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 = some(3'u16)
+  conf.numShardsInNetwork = 1
+  conf.rest = false
+  return conf.toWakuConf().valueOr:
+    raiseAssert error
+
+proc testMessage(): WakuMessage =
+  WakuMessage(
+    payload: "hello".toBytes(),
+    contentTopic: "/test/1/attach/proto",
+    timestamp: 1_700_000_000_000_000_000,
+  )
+
+suite "SendService RLN proof attach":
+  asyncTest "passes the message through unproven when RLN is not mounted":
+    ## The default (no-RLN) configuration must be unaffected: no proof is
+    ## attached and the message reaches the send processors unchanged.
+    let waku = (await Waku.new(testConf())).expect("Waku.new")
+    let msg = testMessage()
+
+    let attached = (await waku.attachRlnProof(msg)).expect("attachRlnProof")
+
+    check:
+      attached.proof.len == 0
+      attached.payload == msg.payload
+      attached.contentTopic == msg.contentTopic
+
+suite "SendService RLN proof attach - RLN mounted":
+  var
+    waku {.threadvar.}: Waku
+    anvilProc {.threadvar.}: Process
+    manager {.threadvar.}: OnchainGroupManager
+
+  asyncSetup:
+    anvilProc = runAnvil(stateFile = some(DEFAULT_ANVIL_STATE_PATH))
+    manager = waitFor setupOnchainGroupManager(deployContracts = false)
+
+    waku = (await Waku.new(testConf())).expect("Waku.new")
+    await waku.node.setRlnValidator(
+      getWakuRlnConfig(
+        manager = manager,
+        userMessageLimit = 20,
+        index = MembershipIndex(1),
+        epochSizeSec = 600,
+      )
+    )
+
+    let credentials = generateCredentials()
+    (
+      waitFor cast[OnchainGroupManager](waku.node.rln.groupManager).register(
+        credentials, UserMessageLimit(20)
+      )
+    ).isOkOr:
+      assert false, "failed to register RLN credentials: " & error
+
+  asyncTeardown:
+    ## The RLN proof-generator provider is registered on the global broker
+    ## context; without stopping RLN it leaks into the next test's setup.
+    try:
+      await waku.node.rln.stop()
+    except Exception:
+      assert false, "failed to stop RLN: " & getCurrentExceptionMsg()
+    stopAnvil(anvilProc)
+
+  asyncTest "attaches a proof":
+    let attached = (await waku.attachRlnProof(testMessage())).expect("attachRlnProof")
+
+    check attached.proof.len > 0
+
+  asyncTest "is idempotent: a message that already carries a proof is untouched":
+    ## Pins the retry contract: the send service re-attaches on every round, so
+    ## re-attaching must neither draw a fresh nonce nor change the bytes —
+    ## otherwise a retried task would resend under a new nullifier.
+    let first = (await waku.attachRlnProof(testMessage())).expect("first attach")
+    let second = (await waku.attachRlnProof(first)).expect("second attach")
+
+    check:
+      first.proof.len > 0
+      second.proof == first.proof

tests/node/test_wakunode_legacy_lightpush.nim

@@ -186,21 +186,24 @@ suite "RLN Proofs as a Lightpush Service":
 
     # The tests below drive `server.legacyLightpushPublish(...)` against the
     # server node. Because `server.wakuLegacyLightPush` is mounted (and no
-    # legacy client is), the call takes the self-request path — it still runs
-    # the full client-side flow (proof gen, retry on RlnValidatorErrorMsg
-    # substring, one-retry cap), but the request lands in the local
+    # legacy client is), the call takes the self-request path — it runs the
+    # full client-side flow (proof gen, RLN-rejection detection via the
+    # RlnValidatorErrorMsg substring), but the request lands in the local
     # pushHandler. Swapping in a stub pushHandler lets each test control what
-    # attempt N sees.
+    # the publish attempt sees.
+    #
+    # On an RLN rejection the publish schedules a background merkle-proof
+    # refresh and returns the error tagged with RlnProofRefreshScheduledMsg;
+    # the caller (the send service loop) regenerates the proof and republishes
+    # on its next round.
 
-    asyncTest "retry fires on RlnValidatorErrorMsg substring and second attempt succeeds":
+    asyncTest "RLN rejection schedules a refresh and surfaces the tagged error":
       var callCount = 0
       let stub: PushMessageHandler = proc(
           pubsubTopic: PubsubTopic, message: WakuMessage
       ): Future[WakuLightPushResult[void]] {.async.} =
         inc callCount
-        if callCount == 1:
-          return err(RlnValidatorErrorMsg & ": simulated stale merkle path")
-        return ok()
+        return err(RlnValidatorErrorMsg & ": simulated stale merkle path")
       server.wakuLegacyLightPush.pushHandler = stub
 
       let response = await server.legacyLightpushPublish(
@@ -208,8 +211,10 @@ suite "RLN Proofs as a Lightpush Service":
       )
 
       check:
-        callCount == 2
-        response.isOk()
+        callCount == 1
+        response.isErr()
+        response.error.contains(RlnProofRefreshScheduledMsg)
+        response.error.contains(RlnValidatorErrorMsg)
 
     asyncTest "no retry when error does not contain RlnValidatorErrorMsg":
       var callCount = 0
@@ -229,27 +234,9 @@ suite "RLN Proofs as a Lightpush Service":
         response.isErr()
         response.error == "unrelated failure"
 
-    asyncTest "retry cap: two consecutive RLN errors surface the second":
-      var callCount = 0
-      let stub: PushMessageHandler = proc(
-          pubsubTopic: PubsubTopic, message: WakuMessage
-      ): Future[WakuLightPushResult[void]] {.async.} =
-        inc callCount
-        return err(RlnValidatorErrorMsg & ": still stale")
-      server.wakuLegacyLightPush.pushHandler = stub
-
-      let response = await server.legacyLightpushPublish(
-        some(pubsubTopic), message, server.peerInfo.toRemotePeerInfo()
-      )
-
-      check:
-        callCount == 2
-        response.isErr()
-        response.error.contains(RlnValidatorErrorMsg)
-
-    asyncTest "no retry when node.rln is nil":
-      # Detach RLN so the retry branch short-circuits on rln.isNone() even
-      # when the error string carries RlnValidatorErrorMsg. Restore before
+    asyncTest "no refresh scheduled when node.rln is nil":
+      # Detach RLN so the RLN-rejection branch short-circuits on rln.isNone()
+      # even when the error string carries RlnValidatorErrorMsg. Restore before
       # teardown so server.stop() sees the same object graph it was
       # constructed with.
       let savedRln = server.rln