feat: attach RLN proofs at the SendService transmission stage

The relay send path published without an RLN proof: proof generation
lived client-side in (legacy)lightpushPublish, so messages dispatched
through SendService -> RelaySendProcessor reached the network unproven
and would be rejected by an RLN-enforcing relay.

Adds Waku.attachRlnProof in the waku/api publish surface and calls it
from SendService immediately after admission, in both send() and the
retry loop. Placement is load-bearing:

- After admit(), so a message rejected by the rate limiter never draws
  a nonce.
- At transmission rather than API entry, because a proof binds to the
  epoch current when the message goes out, and a task can be retried
  for up to MaxTimeInCache after send() returns.

attachRlnProof is a no-op without RLN mounted (message passes through
unproven, as today) and short-circuits on a message that already
carries a proof, so retrying a task neither redraws a nonce nor
changes the bytes. It uses generateRLNProofWithRootRefresh rather than
the plain generator: a task can wait in the 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.

Proof-generation failure parks the task as NextRoundRetry rather than
failing it, matching the admission path: the dominant failure is
NonceLimitReached (RLN's own per-epoch budget exhausted), which the
service loop resolves as the epoch rolls over.

Adds tests/messaging/test_rln_proof_attach.nim covering the unmounted
pass-through, attach when mounted, and the idempotency contract that
the retry loop depends on.

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

logos_delivery/messaging/delivery_service/send_service/send_service.nim

@@ -260,6 +260,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 +305,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/waku/api/publish.nim

@@ -8,7 +8,7 @@
 import logos_delivery/waku/compat/option_valueor
 {.push raises: [].}
 
-import std/options
+import std/[options, times]
 import results, chronos
 
 import logos_delivery/waku/waku
@@ -40,10 +40,34 @@ 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)
+
 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()

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