From 6a1af006b1ed1f1e03218372fece53c4b48c9fc3 Mon Sep 17 00:00:00 2001 From: Tanya S <120410716+stubbsta@users.noreply.github.com> Date: Mon, 3 Aug 2026 10:15:43 +0200 Subject: [PATCH] feat: attach and refresh RLN proofs in the send service (Stream B) (#4069) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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) * 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) * test: cover currentRlnEpochQuota (mounted and unmounted) currentRlnEpochQuota ships in the enforcement PR, but its mounted-RLN assertion needs the anvil-backed group-manager scaffolding that lives in this file, so the coverage rides along here: none when RLN is unmounted, and the epoch index + userMessageLimit when it is. Co-Authored-By: Claude Opus 4.8 * refactor: consolidate the RLN-rejection parking into parkForRlnProofRefresh The relay and lightpush processors duplicated the RLN-rejection recovery (schedule background refresh, clear the stale proof, reset admission, park as NextRoundRetry); both now call parkForRlnProofRefresh in send_processor, so the proof-clear the retry contract depends on cannot drift between the two processors. Also resets firstAdmittedTime so the regenerated proof's fresh nonce is re-admitted rather than sent uncharged. No behavior change. Co-Authored-By: Claude Opus 4.8 * docs: clarify that lightpush reuses an already-attached RLN proof The old wording ("attaches an RLN proof per attempt") reads as if every retry redraws a nonce. The flow proves a message only when it carries no proof, so a task admitted once reuses its proof and nonce across retries. Co-Authored-By: Claude Opus 4.8 (1M context) * feat: retry once on scheduled RLN proof refresh in the lightpush REST handlers The kernel lightpush publish paths no longer retry an RLN-invalid publish inline. On a stale merkle root they schedule a background cache refresh and return early, tagging the error with RlnProofRefreshScheduledMsg — retrying is the caller's decision, so the send service recovers through its own loop and the kernel exposes mechanism only. The synchronous REST endpoints have no such loop: they call publish once and map the result to an HTTP status. Left unchanged, a transient stale-root rejection that the kernel previously absorbed via runRlnRefreshRetry would now surface to the HTTP client as a 503. Restore the transparent retry where it belongs under this layering — at the caller — instead of back in the kernel where it would re-nest inside the send service's retry. Both the legacy and v3 handlers now retry the publish exactly once when the first result carries RlnProofRefreshScheduledMsg, under the same FutTimeoutForPushRequestProcessing bound. The handler's message carries no proof, so the retry regenerates against the refreshed merkle path. Any other error, and any error on the retry itself, maps to its response as before. Co-Authored-By: Claude Opus 4.8 * fix: retry RLN proof attach on a charged-but-unproven task admitAndProve set firstAdmittedTime before attaching the proof, then guarded its whole body on firstAdmittedTime.isSome(). A transient proof attach failure (e.g. NonceLimitReached) left the task charged but with an empty proof, and the next round's early-return skipped the attach entirely and shipped the message bare. The docstring's own invariant — "once admitted, a task keeps its slot and its proof" — was violated: it kept the slot but not the proof. Guard only the rate-limit charge on firstAdmittedTime, not the attach. attachRlnProof is already idempotent (short-circuits when RLN is unmounted or a proof is present), so it is safe to call every round: a charged-but- unproven task retries the attach until it sticks, then short-circuits. The ordering invariant holds (charge strictly before attach, so an over-budget message never draws a nonce), the charge stays once-per-task, and NO_PEERS retries remain free. This also removes a latent relay double-charge: previously a bare message reached the relay, was rejected as RLN-invalid, and parkForRlnProofRefresh reset firstAdmittedTime — re-charging a slot on the next round. The message now never leaves unproven. Co-Authored-By: Claude Opus 4.8 --------- Co-authored-by: Claude Opus 4.8 (1M context) --- .../send_service/lightpush_processor.nim | 5 + .../send_service/relay_processor.nim | 9 ++ .../send_service/send_processor.nim | 16 ++- .../send_service/send_service.nim | 37 ++++-- logos_delivery/waku/api/publish.nim | 60 ++++++++- .../waku/node/waku_node/lightpush.nim | 42 ++----- .../endpoint/legacy_lightpush/handlers.nim | 22 +++- .../rest_api/endpoint/lightpush/handlers.nim | 24 +++- logos_delivery/waku/rln/constants.nim | 5 - logos_delivery/waku/rln/proof.nim | 30 ++++- tests/messaging/test_all.nim | 5 +- tests/messaging/test_rln_proof_attach.nim | 116 ++++++++++++++++++ tests/node/test_wakunode_legacy_lightpush.nim | 49 +++----- 13 files changed, 320 insertions(+), 100 deletions(-) create mode 100644 tests/messaging/test_rln_proof_attach.nim 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 58ffc3d70..ad3aed886 100644 --- a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim +++ b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim @@ -33,6 +33,11 @@ method sendImpl*( await self.waku.lightpushPublishToAny(task.pubsubTopic, task.msg) ).valueOr: error "LightpushSendProcessor.sendImpl failed", error = error.desc.get($error.code) + + if error.isRlnRejection(): + task.parkForRlnProofRefresh(self.waku) + 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 cc53d0fb3..dd61ab676 100644 --- a/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim +++ b/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim @@ -1,6 +1,7 @@ import results, 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] @@ -9,6 +10,7 @@ logScope: topics = "send service relay processor" type RelaySendProcessor* = ref object of BaseSendProcessor + waku: Waku publishProc: PushMessageHandler fallbackStateToSet: DeliveryState @@ -16,6 +18,7 @@ proc new*( T: typedesc[RelaySendProcessor], lightpushAvailable: bool, publishProc: PushMessageHandler, + waku: Waku, brokerCtx: BrokerContext, ): RelaySendProcessor = let fallbackStateToSet = @@ -25,6 +28,7 @@ proc new*( DeliveryState.FailedToDeliver return RelaySendProcessor( + waku: waku, publishProc: publishProc, fallbackStateToSet: fallbackStateToSet, brokerCtx: brokerCtx, @@ -60,6 +64,11 @@ 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(): + task.parkForRlnProofRefresh(self.waku) + 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_processor.nim b/logos_delivery/messaging/delivery_service/send_service/send_processor.nim index ae6d7141d..3c7358d42 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_processor.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_processor.nim @@ -1,5 +1,6 @@ -import chronos +import results, chronos import brokers/broker_context +import logos_delivery/waku/waku, logos_delivery/waku/api/publish import ./delivery_task {.push raises: [].} @@ -21,6 +22,19 @@ method sendImpl*( ): Future[void] {.async, base.} = assert false, "Not implemented" +proc parkForRlnProofRefresh*(task: DeliveryTask, waku: Waku) = + ## The service refused the task's proof as RLN-invalid: the message itself is + ## fine, its proof went stale against a moved merkle root. Schedules a + ## background merkle-path refresh and clears the proof so the next round + ## regenerates one against the refreshed path — `attachRlnProof` + ## short-circuits on an existing proof, so without the clear the rejected + ## bytes would be resent until age-out. Resetting admission re-charges the + ## fresh nonce that regeneration draws. + waku.onRlnProofRejected() + task.msg.proof = @[] + task.firstAdmittedTime = Opt.none(Moment) + task.state = DeliveryState.NextRoundRetry + method process*( self: BaseSendProcessor, task: DeliveryTask ): Future[void] {.async, base.} = 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 18e43a2a3..b18f17d36 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim @@ -68,7 +68,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)) @@ -254,18 +256,27 @@ proc evaluateAndCleanUp(self: SendService) = ) ) -proc admitOnce(self: SendService, task: DeliveryTask): Future[bool] {.async.} = - ## Charges the task's first transmission against the epoch budget, at most - ## once per task (`firstAdmittedTime`); retries then resend for free. Returns - ## false when the task must stay parked for a later epoch. - if task.firstAdmittedTime.isSome(): - return true +proc admitAndProve(self: SendService, task: DeliveryTask): Future[bool] {.async.} = + ## Gates a task's first transmission: charges one epoch slot, then attaches + ## an RLN proof — strictly in that order, so an over-budget message never + ## draws a nonce. The slot is charged at most once per task lifetime + ## (`firstAdmittedTime`); the proof attach is retried each round until it + ## sticks, then short-circuits, so a task charged but not yet proven never + ## ships bare. Returns false while the task must stay parked for a later round. + if task.firstAdmittedTime.isNone(): + (await self.rateLimitManager.admit(task.msg.payload)).isOkOr: + debug "over rate-limit budget, task waits for the epoch to roll", + requestId = task.requestId, msgHash = task.msgHash.to0xHex() + return false + task.firstAdmittedTime = Opt.some(Moment.now()) - (await self.rateLimitManager.admit(task.msg.payload)).isOkOr: - debug "over rate-limit budget, task waits for the epoch to roll", - requestId = task.requestId, msgHash = task.msgHash.to0xHex() + ## A no-op when RLN is not mounted, or when a prior round already attached a + ## proof; otherwise draws the nonce and attaches. + task.msg = (await self.waku.attachRlnProof(task.msg)).valueOr: + error "failed to attach RLN proof, retrying next round", + requestId = task.requestId, error = error return false - task.firstAdmittedTime = Opt.some(Moment.now()) + return true proc trySendMessages*(self: SendService) {.async.} = @@ -273,7 +284,7 @@ proc trySendMessages*(self: SendService) {.async.} = for task in tasksToSend: # Todo, check if it has any perf gain to run them concurrent... - if not (await self.admitOnce(task)): + if not (await self.admitAndProve(task)): continue await self.sendProcessor.process(task) @@ -304,7 +315,7 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} = error "SendService.send: failed to subscribe to content topic", contentTopic = task.msg.contentTopic, error = error - if not (await self.admitOnce(task)): + if not (await self.admitAndProve(task)): info "SendService.send: parking task for a later round", requestId = task.requestId, msgHash = task.msgHash.to0xHex() task.state = DeliveryState.NextRoundRetry diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index 8347c6057..a53f227eb 100644 --- a/logos_delivery/waku/api/publish.nim +++ b/logos_delivery/waku/api/publish.nim @@ -7,6 +7,7 @@ ## so the messaging layer never inspects `waku.node` directly. {.push raises: [].} +import std/[times, strutils] import results, chronos import logos_delivery/waku/waku @@ -38,8 +39,8 @@ 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 currentRlnEpochQuota*(self: Waku): Opt[tuple[epochIndex, messageLimit: uint64]] = @@ -53,6 +54,55 @@ proc currentRlnEpochQuota*(self: Waku): Opt[tuple[epochIndex, messageLimit: uint return Opt.some((fromEpoch(self.node.rln.getCurrentEpoch()), uint64(limit))) +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, Opt.some(shard)).isSome() @@ -61,9 +111,9 @@ proc lightpushPublishToAny*( self: Waku, shard: PubsubTopic, message: WakuMessage ): Future[WakuLightPushResult] {.async.} = ## Selects a lightpush service peer for `shard` and publishes `message` - ## through the node's lightpush flow, which attaches an RLN proof per - ## attempt when RLN is mounted. Returns SERVICE_NOT_AVAILABLE when no peer - ## is available. + ## through the node's lightpush flow. With RLN mounted the flow proves + ## `message` only if it carries no proof, so an already-proven task reuses its + ## nonce. Returns SERVICE_NOT_AVAILABLE when no peer is available. let peer = self.node.peerManager.selectPeer(WakuLightPushCodec, Opt.some(shard)).valueOr: return lightpushResultServiceUnavailable("no lightpush peer available for shard") try: diff --git a/logos_delivery/waku/node/waku_node/lightpush.nim b/logos_delivery/waku/node/waku_node/lightpush.nim index ff47796fc..5a379f7a4 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: Opt[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: Opt[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/rest_api/endpoint/legacy_lightpush/handlers.nim b/logos_delivery/waku/rest_api/endpoint/legacy_lightpush/handlers.nim index a2748de45..80c1ec0ad 100644 --- a/logos_delivery/waku/rest_api/endpoint/legacy_lightpush/handlers.nim +++ b/logos_delivery/waku/rest_api/endpoint/legacy_lightpush/handlers.nim @@ -3,6 +3,7 @@ import std/strformat, std/options, + std/strutils, results, stew/byteutils, chronicles, @@ -15,6 +16,7 @@ import logos_delivery/waku/node/peer_manager, logos_delivery/waku/waku_lightpush_legacy/common, ../../../waku_node, + ../../../rln, ../../handlers, ../serdes, ../responses, @@ -72,13 +74,27 @@ proc installLightPushRequestHandler*( peerOp.valueOr: return NoPeerNoneFoundError - let subFut = node.legacyLightpushPublish(req.pubsubTopic, msg, peer) + var pushFut = node.legacyLightpushPublish(req.pubsubTopic, msg, peer) - if not await subFut.withTimeout(FutTimeoutForPushRequestProcessing): + if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing): error "Failed to request a message push due to timeout!" return RestApiResponse.serviceUnavailable("Push request timed out") - subFut.value().isOkOr: + var pushResult = pushFut.value() + + # An error tagged RlnProofRefreshScheduledMsg is a publish rejected on a + # stale merkle root. On this error the kernel scheduled a cache refresh + # before failing early. This synchronous endpoint has no retry loop of its + # own, so retry once — the second attempt generates its proof against the + # refreshed merkle path. + if pushResult.isErr() and pushResult.error.contains(RlnProofRefreshScheduledMsg): + pushFut = node.legacyLightpushPublish(req.pubsubTopic, msg, peer) + if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing): + error "Failed to request a message push due to timeout!" + return RestApiResponse.serviceUnavailable("Push request timed out") + pushResult = pushFut.value() + + pushResult.isOkOr: if error == TooManyRequestsMessage: return RestApiResponse.tooManyRequests("Request rate limmit reached") diff --git a/logos_delivery/waku/rest_api/endpoint/lightpush/handlers.nim b/logos_delivery/waku/rest_api/endpoint/lightpush/handlers.nim index 18e1eb10a..f96f1c8a3 100644 --- a/logos_delivery/waku/rest_api/endpoint/lightpush/handlers.nim +++ b/logos_delivery/waku/rest_api/endpoint/lightpush/handlers.nim @@ -3,6 +3,7 @@ import std/strformat, std/options, + std/strutils, results, stew/byteutils, chronicles, @@ -15,6 +16,7 @@ import logos_delivery/waku/node/peer_manager, logos_delivery/waku/waku_lightpush/common, ../../../waku_node, + ../../../rln, ../../handlers, ../serdes, ../responses, @@ -96,11 +98,27 @@ proc installLightPushRequestHandler*( makeRestResponse(lightpushResultServiceUnavailable(NoPeerNoneFoundError)) toPeer = Opt.some(aPeer) - let subFut = node.lightpushPublish(req.pubsubTopic, msg, toPeer) + var pushFut = node.lightpushPublish(req.pubsubTopic, msg, toPeer) - if not await subFut.withTimeout(FutTimeoutForPushRequestProcessing): + if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing): error "Failed to request a message push due to timeout!" return makeRestResponse(lightpushResultServiceUnavailable("Push request timed out")) - return makeRestResponse(subFut.value()) + var pushResult = pushFut.value() + + # An error tagged RlnProofRefreshScheduledMsg is a publish rejected on a + # stale merkle root. On this error the kernel scheduled a cache refresh + # before failing early. This synchronous endpoint has no retry loop of its + # own, so retry once — the second attempt generates its proof against the + # refreshed merkle path. + if pushResult.isErr() and + pushResult.error.desc.get("").contains(RlnProofRefreshScheduledMsg): + pushFut = node.lightpushPublish(req.pubsubTopic, msg, toPeer) + if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing): + error "Failed to request a message push due to timeout!" + return + makeRestResponse(lightpushResultServiceUnavailable("Push request timed out")) + pushResult = pushFut.value() + + return makeRestResponse(pushResult) 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 ba4b0361a..b3bbb3bf9 100644 --- a/logos_delivery/waku/rln/proof.nim +++ b/logos_delivery/waku/rln/proof.nim @@ -60,12 +60,14 @@ 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: (raises: []).} = + ## 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) try: let proof = (await rln.groupManager.generateProof(input, epoch, nonce)).valueOr: return err("could not generate rln-v2 proof: " & $error) @@ -73,13 +75,29 @@ proc generateRLNProof*( except CatchableError as e: return err("exception generating rln proof: " & e.msg) +proc generateRLNProof*( + rln: Rln, input: seq[byte], senderEpochTime: float64 +): Future[Result[seq[byte], string]] {.async: (raises: []).} = + 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: @@ -90,7 +108,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/messaging/test_all.nim b/tests/messaging/test_all.nim index c38b65c4e..c3c6b8363 100644 --- a/tests/messaging/test_all.nim +++ b/tests/messaging/test_all.nim @@ -1,4 +1,7 @@ {.used.} import - ./test_rate_limit_manager, ./test_delivery_task_reaping, ./test_send_service_scheduler + ./test_rate_limit_manager, + ./test_rln_proof_attach, + ./test_delivery_task_reaping, + ./test_send_service_scheduler diff --git a/tests/messaging/test_rln_proof_attach.nim b/tests/messaging/test_rln_proof_attach.nim new file mode 100644 index 000000000..11f7d754a --- /dev/null +++ b/tests/messaging/test_rln_proof_attach.nim @@ -0,0 +1,116 @@ +{.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 = Opt.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 + + asyncTest "currentRlnEpochQuota is none when RLN is not mounted": + ## The rate limit manager reads `none` as "use the wall-clock fallback". + let waku = (await Waku.new(testConf())).expect("Waku.new") + check waku.currentRlnEpochQuota().isNone() + +suite "SendService RLN proof attach - RLN mounted": + var + waku {.threadvar.}: Waku + anvilProc {.threadvar.}: Process + manager {.threadvar.}: OnchainGroupManager + + asyncSetup: + anvilProc = runAnvil(stateFile = Opt.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 "currentRlnEpochQuota reports RLN's epoch and user message limit": + ## Wires the rate limit manager to RLN: the manager clamps its configured + ## cap to `messageLimit` and rolls on `epochIndex`. + let quota = waku.currentRlnEpochQuota() + check: + quota.isSome() + quota.get().messageLimit == 20'u64 # the mounted userMessageLimit + quota.get().epochIndex > 0'u64 # unixTime div epochSize, far from zero + + 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 diff --git a/tests/node/test_wakunode_legacy_lightpush.nim b/tests/node/test_wakunode_legacy_lightpush.nim index bbd37fb6e..cc32d5d42 100644 --- a/tests/node/test_wakunode_legacy_lightpush.nim +++ b/tests/node/test_wakunode_legacy_lightpush.nim @@ -187,21 +187,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( @@ -209,8 +212,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 @@ -230,27 +235,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( - Opt.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