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