feat: handle RLN publish rejections in the send service retry loop

An RLN-invalid publish rejection now recovers through the send
service's existing retry loop instead of an inline retry in the kernel.

When a relay or lightpush publish is rejected as RLN-invalid, the
processor clears the message's stale proof, schedules a background
merkle-proof refresh, and parks the task as NextRoundRetry. The next
loop round re-admits the task and regenerates the proof against the
refreshed path. Clearing the proof is required: attachRlnProof
short-circuits on a message that already carries one, so without the
clear the task would resend the rejected proof until it ages out. The
relay processor previously failed such tasks outright, with no
recovery.

Kernel changes supporting this:

- Remove runRlnRefreshRetry from legacyLightpushPublish. The legacy
  path now schedules the refresh and returns the error tagged with
  RlnProofRefreshScheduledMsg, matching the non-legacy path; retrying
  is the caller's decision. Drops the now-unused
  RlnMerkleProofRefreshTimeout.
- generateRLNProofWithRootRefresh reuses the nonce drawn for the first
  attempt when it regenerates after a stale root, rather than drawing a
  second. Only the merkle path differs between the two attempts, so a
  redraw would spend two message ids from the epoch budget on a single
  message and drift the rate limit manager's accounting away from the
  nonce manager's.

Adds Waku.isRlnRejection / Waku.onRlnProofRejected as the messaging
layer's handle on the kernel's RLN rejection detection and background
refresh. Updates the legacy lightpush tests to the schedule-refresh
contract.

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

Author: stubbsta

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,13 +12,15 @@ logScope:
   topics = "send service relay processor"
 
 type RelaySendProcessor* = ref object of BaseSendProcessor
+  waku: Waku
   publishProc: PushMessageHandler
   fallbackStateToSet: DeliveryState
 
 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))
 

logos_delivery/waku/api/publish.nim

@@ -8,7 +8,7 @@
 import logos_delivery/waku/compat/option_valueor
 {.push raises: [].}
 
-import std/[options, times]
+import std/[options, times, strutils]
 import results, chronos
 
 import logos_delivery/waku/waku
@@ -68,6 +68,31 @@ proc attachRlnProof*(
 
   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
-
-    return await runRlnRefreshRetry(
-      node, rln, msgWithProof, pubsubForPublish, peer, 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
+
+    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/node/test_wakunode_legacy_lightpush.nim

@@ -186,30 +186,35 @@ 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.
-
-    asyncTest "retry fires on RlnValidatorErrorMsg substring and second attempt succeeds":
+    # 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 "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(
         some(pubsubTopic), message, server.peerInfo.toRemotePeerInfo()
       )
 
       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