From bed1fc6071b68f22c1fb7714de07a43fe7f25027 Mon Sep 17 00:00:00 2001 From: stubbsta Date: Wed, 15 Jul 2026 12:57:27 +0200 Subject: [PATCH] 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) --- .../send_service/lightpush_processor.nim | 11 +++++ .../send_service/relay_processor.nim | 15 ++++++ .../send_service/send_service.nim | 4 +- logos_delivery/waku/api/publish.nim | 27 +++++++++- .../waku/node/waku_node/lightpush.nim | 42 ++++------------ logos_delivery/waku/rln/constants.nim | 5 -- logos_delivery/waku/rln/proof.nim | 30 +++++++++--- tests/node/test_wakunode_legacy_lightpush.nim | 49 +++++++------------ 8 files changed, 107 insertions(+), 76 deletions(-) diff --git a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim index fa199774b..554b5e166 100644 --- a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim +++ b/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, diff --git a/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim b/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim index dc40b797d..3dd2a3aaa 100644 --- a/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim +++ b/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 diff --git a/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim index a3a59bbaa..c6c774ffd 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/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)) diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index 0c4b5ad03..fb7dfa1ee 100644 --- a/logos_delivery/waku/api/publish.nim +++ b/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() diff --git a/logos_delivery/waku/node/waku_node/lightpush.nim b/logos_delivery/waku/node/waku_node/lightpush.nim index 56d39aae3..ab2c3ef63 100644 --- a/logos_delivery/waku/node/waku_node/lightpush.nim +++ b/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()) diff --git a/logos_delivery/waku/rln/constants.nim b/logos_delivery/waku/rln/constants.nim index 9f93d5fe0..518365b56 100644 --- a/logos_delivery/waku/rln/constants.nim +++ b/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 diff --git a/logos_delivery/waku/rln/proof.nim b/logos_delivery/waku/rln/proof.nim index 8a29eb67b..8402f8cb1 100644 --- a/logos_delivery/waku/rln/proof.nim +++ b/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 diff --git a/tests/node/test_wakunode_legacy_lightpush.nim b/tests/node/test_wakunode_legacy_lightpush.nim index aec37e18c..69cc3e269 100644 --- a/tests/node/test_wakunode_legacy_lightpush.nim +++ b/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