From 77cb8a6c7a199b7823e5149bb72f8ce384883b8f Mon Sep 17 00:00:00 2001 From: Tanya S <120410716+stubbsta@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:27:37 +0200 Subject: [PATCH] refresh merkle proof cache reactively on msg publish rejection (#4013) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refresh merkle proof cache reactively on lightpush rejection * Add tests for lightpush force refresh regenerates proof * Fix linting * Add nonce rollback for lightpush publish retry * Fix nonce rollback race: make rollbackNonce CAS-conditional Two bugs in the previous blind-decrement rollbackNonce(): 1. Concurrent draw: another proof generated between attempt 1 and the retry advances nextNonce, so decrementing steals a nonce already claimed by another live message — nullifier collision, on-chain slashable. 2. Pre-attached proof: if the incoming message already carries a proof, checkAndGenerateRLNProof short-circuits and draws no nonce, yet the retry still called rollback, stomping on whatever nonce the preceding message legitimately drew. Fix: rollbackNonce now takes the specific nonce as a parameter and only decrements when nextNonce == nonce + 1. checkAndGenerateRLNProof returns (msg, messageId) so callers can thread the drawn id through to the retry as reuseMessageId; generateRLNProof no longer performs an implicit rollback. Co-Authored-By: Claude Sonnet 4.6 * Extend reactive Merkle proof refresh to REST relay and broker paths With PathCheckMinInterval removed, a non-empty merkleProofCache is trusted forever unless force=true is passed. Previously only the two lightpush retry paths ever passed force=true; the REST relay publish handlers (static and auto-sharding) and the broker proof provider had no rejection feedback loop, so a sliding root window would permanently reject their messages until restart. - REST relay handlers: run validateMessage first; if it returns RlnValidatorErrorMsg, force-refresh the cached path and retry validation once before publishing — mirroring the lightpush reactive pattern. - Broker provider (rln.nim): decode the generated proof bytes, call validateRoot on the embedded Merkle root, and force-refresh + regenerate when the root is outside the acceptance window — since the broker has no external rejection signal to react to. Tests added for all three new paths, using the corrupted-cache technique (all-zero merkleProofCache → garbage root → rejection → retry). Co-Authored-By: Claude Sonnet 4.6 * Gate lightpush Merkle refresh on RlnValidatorErrorMsg for 420 responses 420 INVALID_MESSAGE is returned for any validateMessage failure, not just RLN ones (e.g. oversized messages), so a bare status-code check triggers an unbounded on-chain fetchMerkleProofElements RPC per rejected message. Match the legacy-lightpush path: require the error description to contain RlnValidatorErrorMsg before refreshing; 504 OUT_OF_RLN_PROOF remains unconditional as it is unambiguously RLN-specific. Co-Authored-By: Claude Sonnet 4.6 * Fix merge conflict residue: replace RlnResult with Result in proof.nim RlnResult[T] was removed as a type alias in commit 0a1700e23 (replace RLN specific Result with generic Result). The merge left one stale reference at proof.nim:62; replace with the equivalent Result[seq[byte], string]. Co-Authored-By: Claude Sonnet 4.6 * Fix ResultDefect on success and stale test stubs in lightpush retry path Two bugs left by a bad merge: 1. lightpush.nim: isRlnRelatedFailure was computed before firstResult.isOk() was checked, accessing firstResult.error unconditionally. When publish succeeds, this raises a ResultDefect. Reorder the guard so isOk()/isNone() short-circuits before the error fields are touched. 2. test_wakunode_lightpush.nim: the 420-retry stubs used descriptions ("simulated stale merkle path", "still stale") that don't contain RlnValidatorErrorMsg. The code gates 420 retries on that substring to distinguish RLN rejections from unrelated INVALID_MESSAGE responses (e.g. oversized messages). Update stubs to emit RlnValidatorErrorMsg so the retry path fires as the tests expect. Co-Authored-By: Claude Sonnet 4.6 * Fix linting * Dedupe checkAndGenerateRLNProof force-refresh test Same test existed in both test_wakunode_lightpush.nim and test_wakunode_legacy_lightpush.nim. It exercises the checkAndGenerateRLNProof primitive (short-circuit override on force=true + nonce rollback + cache refetch), not lightpush-specific behavior. The modern lightpush copy provides full coverage; the legacy copy paid the anvil/on-chain setup cost for zero incremental coverage. Co-Authored-By: Claude Sonnet 4.6 * Bound lightpush RLN refresh-retry with a 5s timeout The reactive refresh + retry path calls into the RLN contract (eth_call for getMerkleProof) then republishes over libp2p — neither has an outer bound, so a hanging RPC endpoint could stall the caller indefinitely. Wrap the retry in withTimeout(RlnRefreshRetryTimeout); on timeout, log and return the original error rather than hang. Applied to both the legacy and modern lightpushPublish retry paths. Co-Authored-By: Claude Opus 4.7 * Extract broker RLN proof provider into generateRLNProofWithRootRefresh The broker proof provider's inline logic — generate proof, decode it, self-validate the root against the acceptable-root window, and on stale root force-refresh + regenerate — is a reusable primitive on its own. Lift it into rln/proof.nim next to the existing generateRLNProof so the provider closure in mount() collapses to a single call, and any future non-broker caller can share the same flow. Pure refactor; behavior identical. Co-Authored-By: Claude Opus 4.7 * Extract REST relay handlers' RLN attach+validate+retry into a helper Both the static and auto POST handlers duplicated the same block: attach an RLN proof, run wakuRelay.validateMessage, and on RlnValidatorErrorMsg substring force-refresh the merkle path and revalidate. Lift the block into attachRlnProofValidateWithRetry alongside a small RlnPublishError variant so the handlers can keep dispatching 500 for proof-gen failures and 400 for validator rejections. Each handler is now ~15 lines instead of ~35 and the retry contract lives in one place. Pure refactor; behavior unchanged. Verified against tests/wakunode_rest/test_rest_relay.nim (13/13) including the two stale- RLN-proof retry tests added earlier on this branch. Co-Authored-By: Claude Opus 4.7 * Simplify legacy lightpush autosharding path to a direct getShard call The single-content-topic path went through getShardsFromContentTopics, iterated the resulting table under the assumption it was 1-entry, and had a defensive fallthrough (`resolved = false` → bare `return`) for the empty-table case. That case is unreachable: on success the table always has exactly one entry, and the fallthrough returned a default- initialized Result — not behavior worth preserving. Replace with the same two-step pattern the modern lightpushPublish uses (NsContentTopic.parse → sharding.getShard) so both handlers share a shape and the (now-explicit) failure paths surface useful error strings. Verified against tests/node/test_wakunode_legacy_lightpush.nim (9/9). Co-Authored-By: Claude Opus 4.7 * Break up legacyLightpushPublish into single-purpose helpers legacyLightpushPublish had grown to ~100 lines with three concerns interleaved: choosing between client/self-request publish, resolving the effective pubsub topic, and running the RLN refresh-retry with timeout. Lift each into a private top-level proc so the main proc reads top-to- bottom in ~35 lines: precondition → prepare message + proof → resolve topic → publish once → RLN-refresh retry if applicable. The new helpers are: - internalLegacyLightpushPublish — client vs self-request dispatch (was the inline closure) - resolveLegacyPubsubTopic — explicit param or autosharding - runRlnRefreshRetry — force-refresh proof + retry with RlnRefreshRetryTimeout bound; returns caller's fallback on timeout Nonce now appears in a top-level signature, so import it directly from rln/nonce_manager rather than re-exporting through rln/proof.nim (the latter would leak nonce_manager's bulk chronos/times exports and clash with waku_rendezvous overload resolution). Pure refactor; behavior unchanged. Verified against tests/node/test_wakunode_legacy_lightpush.nim (9/9) and tests/node/test_wakunode_lightpush.nim (11/11). Co-Authored-By: Claude Opus 4.7 * Replace forceMerkleProofRefresh flag with invalidateMerkleProofCache The forceMerkleProofRefresh bool was threaded through five procs (checkAndGenerateRLNProof → generateRLNProof → generateProof base + on-chain → ensureFreshMerkleProofPath) purely so a caller could ask the merkle path to be refetched from chain on a stale-RLN retry. Reading any one of those signatures required a mental stack trace back up to the retry caller to know what "force" meant. Flip the contract: expose invalidateMerkleProofCache on the group manager; callers invoke it on the retry path to empty the cache, and the normal proof-gen flow then sees an empty cache and refetches on its own. No flag crosses procs. Removed forceMerkleProofRefresh from: - group_manager_base.generateProof (+ on_chain override) - ensureFreshMerkleProofPath (its `force` param) - generateRLNProof Renamed on checkAndGenerateRLNProof: forceMerkleProofRefresh → regenerate. It now controls one thing — bypassing the "message already has a proof" short-circuit — and never propagates downstream. Callers pair it with invalidateMerkleProofCache when they need the refetch semantics too. Retry callers updated in lockstep: generateRLNProofWithRootRefresh, REST attachRlnProofValidateWithRetry, and both legacy + modern lightpush retry paths now call groupManager.invalidateMerkleProofCache() before regenerating. The primitive test in test_wakunode_lightpush.nim now exercises the pair directly, and the onchain ensureFreshMerkleProofPath test is rewritten to exercise invalidate → ensureFresh as one flow (retiring its `force = true` variant). Pure refactor; behavior unchanged. Verified against: tests/waku_rln_relay/test_rln_group_manager_onchain.nim (29/29) tests/wakunode_rest/test_rest_relay.nim (13/13) tests/waku_rln_relay/test_wakunode_rln_relay.nim (5/5, 1 skip) tests/node/test_wakunode_lightpush.nim (11/11) tests/node/test_wakunode_legacy_lightpush.nim (9/9) Co-Authored-By: Claude Opus 4.7 * Reject empty merkle proof cache in on-chain generateProof An empty cache passes the existing mod-32 length check and silently produces a witness with no path elements, yielding a proof no validator will accept. Catch the case explicitly so the failure surfaces at the callsite rather than after a wasted publish round-trip. Co-Authored-By: Claude Opus 4.7 * Drop nonce reuse on RLN retry; consume a fresh nonce instead The retry path threaded a rejected attempt's message id back so NonceManager.rollbackNonce could CAS the counter and reclaim the slot. This bought at most one nonce per retry (a rare event) at the cost of an Option[Nonce] param, a tuple return, and a concurrent-draw race the CAS had to defend against. Simpler to optimistically waste the nonce: checkAndGenerateRLNProof returns Result[WakuMessage, string] and callers drop the tuple destructuring. NonceManager.rollbackNonce and its tests, now dead, are removed. Co-Authored-By: Claude Opus 4.7 * Split checkAndGenerateRLNProof: extract attachRLNProof The `regenerate: bool` param was control coupling: every caller passing `regenerate = true` was a retry path that had already established `rln.isSome()` (it just called `invalidateMerkleProofCache` on the group manager) and wanted an unconditional fresh proof, not a "check and maybe generate". Extract that into `attachRLNProof(r: Rln, message)`, which always draws a fresh nonce and rebuilds the proof, replacing any existing one. `checkAndGenerateRLNProof(rln: Option[Rln], message)` keeps only its "check" behaviour — short-circuit on an existing proof, pass through when RLN is not configured — and delegates to `attachRLNProof`. Taking a concrete `Rln` deletes the previously unreachable `regenerate = true` + `rln.isNone()` branch (which would have republished the stale rejected proof) at the type level, and removes the duplicated nonce-draw/generateProof/encode body that checkAndGenerateRLNProof carried alongside generateRLNProof. The two lightpush retry sites and the invalidate+regenerate test now call attachRLNProof directly. Co-Authored-By: Claude Fable 5 * Bound only the proof refresh on lightpush RLN retry, not the republish RlnRefreshRetryTimeout previously wrapped the whole retry leg: merkle path refetch + proof regeneration + the second publish. Sharing one clock meant a slow (but successful) refetch could leave no budget for the publish, cancelling it mid-flight — burning the freshly drawn nonce and, worse, reporting the original RLN rejection while the retried request may already have reached the service node and been relayed. Move the withTimeout to cover only attachRLNProof (the on-chain eth_call + proof regeneration — the one step with an unbounded external dependency). The retried publish runs unbounded after it, matching the first attempt's semantics, so slow transports (e.g. an optional mix route) cannot be cancelled by the refresh budget and the caller always receives the publish's actual result. On refresh timeout the retry publish never starts and the original rejection is returned, so a "failed" answer now guarantees no second request is in flight. Co-Authored-By: Claude Fable 5 * Shield the shared merkle path refetch from caller cancellation ensureFreshMerkleProofPath coalesces concurrent callers onto a single in-flight refetch future and awaited it directly. Chronos propagates cancellation into the awaited future, so one caller being cancelled (e.g. a publish retry whose RlnRefreshRetryTimeout fired mid-refetch) cancelled the shared future under every coalesced caller — and threw away an eth_call that may have been about to complete. Await the in-flight future via join() instead: cancelling a caller only detaches its callback. The refetch always runs to completion, coalesced callers keep waiting, and even when every caller times out the cache still gets populated in the background so the next publish fast-paths instead of paying the RPC round-trip again. Detaching is safe because doRefresh cannot fail with an unhandled exception: the eth_call wrappers catch CatchableError and return Result, and failures surface to callers through fetchOk / an empty cache exactly as before. Adds a test cancelling the initiating caller mid-refetch and asserting the shared future survives and the coalesced caller completes with a populated cache. Co-Authored-By: Claude Fable 5 * Move message publish retries out of the kernel on stale RLN proof On a stale-merkle-proof rejection the kernel and REST layers no longer republish internally. They schedule a detached merkle proof refresh and fail fast with a distinguishable, retry-worthy error, leaving the retry decision to the caller. - add GroupManager.scheduleMerkleProofRefresh: drops the cached path and asyncSpawns the refetch (join-shielded, so a cancelled caller never cancels the shared refetch); base is a no-op. - v3 lightpush (node.lightpushPublish): on an RLN-related 420/504, call scheduleMerkleProofRefresh and return immediately, normalized to 504 OUT_OF_RLN_PROOF with RlnProofRefreshScheduledMsg. The internal republish and its bounded timeout are gone. - send service: lightpushPublishToAny now routes through node.lightpushPublish so every NextRoundRetry round regenerates the proof against the current cache; the existing OUT_OF_RLN_PROOF -> NextRoundRetry mapping is the retry. - REST lightpush: pass the 504 through unchanged (no internal retry). - REST relay: attachRlnProofValidateWithRetry -> attachRlnProofAndValidate; validate once, on RLN-invalid schedule the refresh and return 503 with the marker instead of regenerating and revalidating in-request. - rename RlnRefreshRetryTimeout -> RlnMerkleProofRefreshTimeout; it now bounds only the legacy lightpush refresh, which keeps its one-shot internal retry (no wire error taxonomy to delegate to callers). Not touched: legacy lightpush retry behavior, the proof.nim attachRLNProof/checkAndGenerateRLNProof split, and the merkle path refetch/coalescing internals of ensureFreshMerkleProofPath. Co-Authored-By: Claude Fable 5 * Tighten RLN proof-refresh docstrings and comments Condense the verbose comments introduced with the reactive merkle proof-refresh work. No code changes. - group_manager_base: base methods carry the contract; the OnchainGroupManager overrides keep only their concrete behavior instead of repeating the rationale. - proof.nim, lightpush.nim, constants.nim: trim multi-line docstrings and inline comments to their load-bearing facts. Co-Authored-By: Claude Fable 5 * Make the merkle proof path refresh safe against concurrent invalidation ensureFreshMerkleProofPath returns the fetched path instead of leaving callers to re-read merkleProofCache. generateProof uses the returned value, so an invalidate landing during the refresh's updateMemberCache suspension can no longer empty the cache under an in-flight proof-gen and trip its empty-cache guard. The cache field is now only a hint for the next call. invalidateMerkleProofCache also bumps merkleProofCacheGeneration. The refresh stamps the generation before each fetchMerkleProofElements and only caches a result whose generation is unchanged, otherwise it fetches again, bounded by MerkleProofRefetchMaxAttempts. A rejection-driven invalidate is therefore never absorbed by a fetch that predates it, so the path a caller receives always postdates the invalidate. refreshRoots awaits rootsRefreshInFlightFut via join() at both await sites, giving the roots refresh the same cancellation shield the proof path refetch already had: a cancelled validateRoot caller no longer cancels the refresh its coalesced peers are waiting on. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Sonnet 4.6 --- logos_delivery/waku/api/publish.nim | 9 +- .../waku/node/waku_node/lightpush.nim | 146 +++++++++---- logos_delivery/waku/node/waku_node/relay.nim | 2 +- .../waku/rest_api/endpoint/relay/handlers.nim | 91 +++++--- logos_delivery/waku/rln/constants.nim | 18 +- .../rln/group_manager/group_manager_base.nim | 10 + .../group_manager/on_chain/group_manager.nim | 107 ++++++---- logos_delivery/waku/rln/proof.nim | 44 +++- logos_delivery/waku/rln/rln.nim | 7 +- tests/node/test_wakunode_legacy_lightpush.nim | 97 ++++++++- tests/node/test_wakunode_lightpush.nim | 183 +++++++++++++++- .../test_rln_group_manager_onchain.nim | 195 +++++++++++++++-- .../test_wakunode_rln_relay.nim | 55 ++++- tests/wakunode_rest/test_rest_relay.nim | 198 +++++++++++++++++- 14 files changed, 1011 insertions(+), 151 deletions(-) diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index e45eff942..f9f3229f6 100644 --- a/logos_delivery/waku/api/publish.nim +++ b/logos_delivery/waku/api/publish.nim @@ -16,6 +16,7 @@ import logos_delivery/waku/[ waku_core, node/waku_node, + node/waku_node/lightpush, node/peer_manager, waku_relay/protocol, rln, @@ -50,11 +51,13 @@ proc lightpushPeerAvailable*(self: Waku, shard: PubsubTopic): bool = proc lightpushPublishToAny*( self: Waku, shard: PubsubTopic, message: WakuMessage ): Future[WakuLightPushResult] {.async.} = - ## Selects a lightpush service peer for `shard` and publishes `message`. - ## Returns SERVICE_NOT_AVAILABLE when no peer is available. + ## 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. let peer = self.node.peerManager.selectPeer(WakuLightPushCodec, some(shard)).valueOr: return lightpushResultServiceUnavailable("no lightpush peer available for shard") try: - return await self.node.wakuLightpushClient.publish(some(shard), message, peer) + return await self.node.lightpushPublish(some(shard), message, some(peer)) except CatchableError as e: return lightpushResultInternalError(e.msg) diff --git a/logos_delivery/waku/node/waku_node/lightpush.nim b/logos_delivery/waku/node/waku_node/lightpush.nim index 1bb4ddcbb..56d39aae3 100644 --- a/logos_delivery/waku/node/waku_node/lightpush.nim +++ b/logos_delivery/waku/node/waku_node/lightpush.nim @@ -2,7 +2,7 @@ import logos_delivery/waku/compat/option_valueor {.push raises: [].} import - std/[hashes, options, tables, net], + std/[hashes, options, strutils, tables, net], chronos, chronicles, metrics, @@ -68,6 +68,67 @@ proc mountLegacyLightPushClient*(node: WakuNode) = node.wakuLegacyLightpushClient = WakuLegacyLightPushClient.new(node.peerManager, node.rng) +proc internalLegacyLightpushPublish( + node: WakuNode, pubsubTopic: PubsubTopic, message: WakuMessage, peer: RemotePeerInfo +): Future[legacy_lightpush_protocol.WakuLightPushResult[string]] {.async, gcsafe.} = + ## Dispatches to the legacy lightpush client if mounted, otherwise to the + ## self-hosted server. Callers guarantee at least one is mounted. + let msgHash = pubsubTopic.computeMessageHash(message).to0xHex() + if not node.wakuLegacyLightpushClient.isNil(): + notice "publishing message with legacy lightpush", + pubsubTopic = pubsubTopic, + contentTopic = message.contentTopic, + target_peer_id = peer.peerId, + msg_hash = msgHash + return await node.wakuLegacyLightpushClient.publish(pubsubTopic, message, peer) + + notice "publishing message with self hosted legacy lightpush", + pubsubTopic = pubsubTopic, + contentTopic = message.contentTopic, + target_peer_id = peer.peerId, + msg_hash = msgHash + return await node.wakuLegacyLightPush.handleSelfLightPushRequest(pubsubTopic, message) + +proc resolveLegacyPubsubTopic( + node: WakuNode, pubsubTopic: Option[PubsubTopic], contentTopic: ContentTopic +): Result[PubsubTopic, string] = + ## Returns the explicit pubsub topic, else derives it from `contentTopic` + ## via autosharding. The legacy wire format requires a pubsub topic and the + ## server never derives it, so the client must resolve it here. + if pubsubTopic.isSome(): + return ok(pubsubTopic.get()) + if node.wakuAutoSharding.isNone(): + return err("Pubsub topic must be specified when static sharding is enabled") + let parsedTopic = NsContentTopic.parse(contentTopic).valueOr: + return err("Invalid content-topic: " & $error) + let shard = node.wakuAutoSharding.get().getShard(parsedTopic).valueOr: + 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], @@ -82,9 +143,8 @@ proc legacyLightpushPublish*( error "failed to publish message as legacy lightpush not available" return err("Waku lightpush not available") - # toRLNSignal includes the timestamp in the proof input, so the timestamp - # must be fixed before proof generation. The downstream ensureTimestampSet - # in the client publish becomes an idempotent no-op safety net. + # toRLNSignal hashes the timestamp into the proof, so fix it before proof gen; + # the downstream ensureTimestampSet then becomes a no-op. let message = ensureTimestampSet(message) let rln = @@ -95,40 +155,24 @@ proc legacyLightpushPublish*( let msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: return err("failed call checkAndGenerateRLNProof from lightpush: " & error) - let internalPublish = proc( - node: WakuNode, - pubsubTopic: PubsubTopic, - message: WakuMessage, - peer: RemotePeerInfo, - ): Future[legacy_lightpush_protocol.WakuLightPushResult[string]] {.async, gcsafe.} = - let msgHash = pubsubTopic.computeMessageHash(message).to0xHex() - if not node.wakuLegacyLightpushClient.isNil(): - notice "publishing message with legacy lightpush", - pubsubTopic = pubsubTopic, - contentTopic = message.contentTopic, - target_peer_id = peer.peerId, - msg_hash = msgHash - return await node.wakuLegacyLightpushClient.publish(pubsubTopic, message, peer) - - if not node.wakuLegacyLightPush.isNil(): - notice "publishing message with self hosted legacy lightpush", - pubsubTopic = pubsubTopic, - contentTopic = message.contentTopic, - target_peer_id = peer.peerId, - msg_hash = msgHash - return - await node.wakuLegacyLightPush.handleSelfLightPushRequest(pubsubTopic, message) try: - if pubsubTopic.isSome(): - return await internalPublish(node, pubsubTopic.get(), msgWithProof, peer) + let pubsubForPublish = resolveLegacyPubsubTopic( + node, pubsubTopic, message.contentTopic + ).valueOr: + return err(error) - if node.wakuAutoSharding.isNone(): - return err("Pubsub topic must be specified when static sharding is enabled") - let topicMap = - ?node.wakuAutoSharding.get().getShardsFromContentTopics(message.contentTopic) + let firstResult = + await internalLegacyLightpushPublish(node, pubsubForPublish, msgWithProof, peer) - for pubsub, _ in topicMap.pairs: # There's only one pair anyway - return await internalPublish(node, $pubsub, 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 + ) except CatchableError: return err(getCurrentExceptionMsg()) @@ -280,9 +324,8 @@ proc lightpushPublish*( error "lightpush publish error", error = msg return lighpushErrorResult(LightPushErrorCode.INTERNAL_SERVER_ERROR, msg) - # toRLNSignal includes the timestamp in the proof input, so the timestamp - # must be fixed before proof generation. The downstream ensureTimestampSet - # in the client publish becomes an idempotent no-op safety net. + # toRLNSignal hashes the timestamp into the proof, so fix it before proof gen; + # the downstream ensureTimestampSet then becomes a no-op. let message = ensureTimestampSet(message) let rln = @@ -293,5 +336,30 @@ proc lightpushPublish*( let msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: return lighpushErrorResult(LightPushErrorCode.OUT_OF_RLN_PROOF, error) - return + let firstResult = await lightpushPublishHandler(node, pubsubForPublish, msgWithProof, toPeer, mixify) + + # Gate the refresh on unambiguously RLN-related failures: 504 + # (OUT_OF_RLN_PROOF) is always RLN; 420 (INVALID_MESSAGE) also covers non-RLN + # rejections (e.g. oversized), so additionally require RlnValidatorErrorMsg. + if firstResult.isOk() or rln.isNone(): + return firstResult + let isRlnRelatedFailure = + firstResult.error.code == LightPushErrorCode.OUT_OF_RLN_PROOF or ( + firstResult.error.code == LightPushErrorCode.INVALID_MESSAGE and + firstResult.error.desc.get("").contains(RlnValidatorErrorMsg) + ) + if not isRlnRelatedFailure: + return firstResult + + # Schedule the refresh and return immediately, normalized to 504 with + # RlnProofRefreshScheduledMsg so callers can tell "stale proof, retry" from a + # permanent rejection. A retry regenerates against the refreshed cache. + info "lightpush send rejected as RLN-invalid; scheduling merkle proof refresh", + statusCode = $firstResult.error.code + rln.get().groupManager.scheduleMerkleProofRefresh() + return lighpushErrorResult( + LightPushErrorCode.OUT_OF_RLN_PROOF, + RlnProofRefreshScheduledMsg & ": " & + firstResult.error.desc.get($firstResult.error.code), + ) diff --git a/logos_delivery/waku/node/waku_node/relay.nim b/logos_delivery/waku/node/waku_node/relay.nim index 18a3f0f59..77143fabb 100644 --- a/logos_delivery/waku/node/waku_node/relay.nim +++ b/logos_delivery/waku/node/waku_node/relay.nim @@ -261,4 +261,4 @@ proc setRlnValidator*( # register rln validator as default validator info "Registering RLN validator" - node.wakuRelay.addValidator(validator, "RLN validation failed") + node.wakuRelay.addValidator(validator, RlnValidatorErrorMsg) diff --git a/logos_delivery/waku/rest_api/endpoint/relay/handlers.nim b/logos_delivery/waku/rest_api/endpoint/relay/handlers.nim index 6b1fdcc69..0976877f4 100644 --- a/logos_delivery/waku/rest_api/endpoint/relay/handlers.nim +++ b/logos_delivery/waku/rest_api/endpoint/relay/handlers.nim @@ -51,6 +51,49 @@ proc validatePubSubTopics(topics: seq[PubsubTopic]): Result[void, RestApiRespons return ok() +type + RlnPublishErrorKind = enum + ProofGenFailed ## Local proof generation failed — server-side (500). + ValidationRejected ## Validator rejected the message — client-side (400). + StaleProofSuspected ## Stale merkle path; refresh scheduled — retry (503). + + RlnPublishError = object + kind: RlnPublishErrorKind + desc: string + +proc attachRlnProofAndValidate( + rln: Rln, wakuRelay: WakuRelay, pubsubTopic: PubsubTopic, message: WakuMessage +): Future[Result[WakuMessage, RlnPublishError]] {.async.} = + ## Attaches an RLN proof to `message` and validates it via `wakuRelay`. + ## If the validator rejects it as RLN-invalid (error contains + ## RlnValidatorErrorMsg), schedules a background merkle proof refresh and + ## fails early with StaleProofSuspected — the caller decides whether to + ## retry. Callers invoke only when RLN is mounted. + var msg = message + msg.proof = ( + await rln.generateRLNProof(msg.toRLNSignal(), float64(getTime().toUnix())) + ).valueOr: + return err( + RlnPublishError( + kind: ProofGenFailed, desc: "error appending RLN proof to message: " & $error + ) + ) + + let validateResult = await wakuRelay.validateMessage(pubsubTopic, msg) + if validateResult.isOk(): + return ok(msg) + if not validateResult.error.contains(RlnValidatorErrorMsg): + return err(RlnPublishError(kind: ValidationRejected, desc: validateResult.error)) + + info "relay publish rejected as RLN-invalid; scheduling merkle proof refresh" + rln.groupManager.scheduleMerkleProofRefresh() + return err( + RlnPublishError( + kind: StaleProofSuspected, + desc: RlnProofRefreshScheduledMsg & ": " & validateResult.error, + ) + ) + proc installRelayApiHandlers*( router: var RestRouter, node: WakuNode, cache: MessageCache ) = @@ -166,21 +209,20 @@ proc installRelayApiHandlers*( var message: WakuMessage = reqWakuMessage.toWakuMessage(version = 0).valueOr: return RestApiResponse.badRequest($error) - # if RLN is mounted, append the proof to the message if not node.rln.isNil(): - # append the proof to the message - - message.proof = ( - await node.rln.generateRLNProof( - message.toRLNSignal(), float64(getTime().toUnix()) - ) + message = ( + await attachRlnProofAndValidate(node.rln, node.wakuRelay, pubsubTopic, message) ).valueOr: - return RestApiResponse.internalServerError( - "Failed to publish: error appending RLN proof to message: " & $error - ) - - (await node.wakuRelay.validateMessage(pubsubTopic, message)).isOkOr: - return RestApiResponse.badRequest("Failed to publish: " & error) + case error.kind + of ProofGenFailed: + return RestApiResponse.internalServerError("Failed to publish: " & error.desc) + of ValidationRejected: + return RestApiResponse.badRequest("Failed to publish: " & error.desc) + of StaleProofSuspected: + return RestApiResponse.serviceUnavailable("Failed to publish: " & error.desc) + else: + (await node.wakuRelay.validateMessage(pubsubTopic, message)).isOkOr: + return RestApiResponse.badRequest("Failed to publish: " & error) # Log for message tracking purposes logMessageInfo(node.wakuRelay, "rest", pubsubTopic, "none", message, onRecv = true) @@ -297,19 +339,20 @@ proc installRelayApiHandlers*( error "publish error", err = msg return RestApiResponse.badRequest("Failed to publish. " & msg) - # if RLN is mounted, append the proof to the message if not node.rln.isNil(): - message.proof = ( - await node.rln.generateRLNProof( - message.toRLNSignal(), float64(getTime().toUnix()) - ) + message = ( + await attachRlnProofAndValidate(node.rln, node.wakuRelay, pubsubTopic, message) ).valueOr: - return RestApiResponse.internalServerError( - "Failed to publish: error appending RLN proof to message: " & error - ) - - (await node.wakuRelay.validateMessage(pubsubTopic, message)).isOkOr: - return RestApiResponse.badRequest("Failed to publish: " & error) + case error.kind + of ProofGenFailed: + return RestApiResponse.internalServerError("Failed to publish: " & error.desc) + of ValidationRejected: + return RestApiResponse.badRequest("Failed to publish: " & error.desc) + of StaleProofSuspected: + return RestApiResponse.serviceUnavailable("Failed to publish: " & error.desc) + else: + (await node.wakuRelay.validateMessage(pubsubTopic, message)).isOkOr: + return RestApiResponse.badRequest("Failed to publish: " & error) # Log for message tracking purposes logMessageInfo(node.wakuRelay, "rest", pubsubTopic, "none", message, onRecv = true) diff --git a/logos_delivery/waku/rln/constants.nim b/logos_delivery/waku/rln/constants.nim index 22454b22b..9f93d5fe0 100644 --- a/logos_delivery/waku/rln/constants.nim +++ b/logos_delivery/waku/rln/constants.nim @@ -14,14 +14,22 @@ const RlnContractRootCacheSize* = 5 # Using Linea block generation time as reference, which is around 2 seconds const RootsRefreshMinInterval* = 2.seconds -# Minimum time between two consecutive merkle proof path freshness checks. -# Bounds how often the publish path queries chain when generating proofs at a high rate. -# Using Linea block generation time ~2s and AcceptableRootWindowSize=50, we give a generous safety margin within this -const PathCheckMinInterval* = 30.seconds - # RLN membership key and index files path const RlnCredentialsFilename* = "rlnCredentials.txt" +# RLN Validator message rejection Error string, is used to trigger proof refresh and publish retry in the lightpush client +const RlnValidatorErrorMsg* = "RLN validation failed" + +# OUT_OF_RLN_PROOF description marker telling callers a background refresh was +# scheduled and retrying the publish is worthwhile. +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/group_manager/group_manager_base.nim b/logos_delivery/waku/rln/group_manager/group_manager_base.nim index 21571aa52..5f91cf19f 100644 --- a/logos_delivery/waku/rln/group_manager/group_manager_base.nim +++ b/logos_delivery/waku/rln/group_manager/group_manager_base.nim @@ -141,5 +141,15 @@ method generateProof*( ## Dummy implementation for generateProof return err("generateProof is not implemented") +method invalidateMerkleProofCache*(g: GroupManager) {.base, gcsafe, raises: [].} = + ## Drops the cached merkle proof path so the next proof-gen refetches from + ## chain. Called after a publish is rejected on a stale cache. No-op base. + discard + +method scheduleMerkleProofRefresh*(g: GroupManager) {.base, gcsafe, raises: [].} = + ## Like `invalidateMerkleProofCache`, but starts the refetch in the + ## background so the caller need not wait for it. No-op base. + discard + method isReady*(g: GroupManager): Future[bool] {.base, async.} = return true diff --git a/logos_delivery/waku/rln/group_manager/on_chain/group_manager.nim b/logos_delivery/waku/rln/group_manager/on_chain/group_manager.nim index e316de85b..a912306a5 100644 --- a/logos_delivery/waku/rln/group_manager/on_chain/group_manager.nim +++ b/logos_delivery/waku/rln/group_manager/on_chain/group_manager.nim @@ -44,8 +44,8 @@ type registrationHandler*: Option[RegistrationHandler] latestProcessedBlock*: BlockNumber merkleProofCache*: seq[byte] - lastMerklePathCheckMoment*: Moment - proofPathRefreshInFlightFut*: Future[void] + merkleProofCacheGeneration: uint64 + proofPathRefreshInFlightFut*: Future[seq[byte]] lastRootsRefreshMoment*: Moment rootsRefreshInFlightFut*: Future[void] @@ -236,12 +236,13 @@ proc updateMemberCount*( proc refreshRoots(g: OnchainGroupManager): Future[void] {.async.} = ## On-demand refresh of validRoots from the on-chain root cache. ## Throttled to at most one refresh per RootsRefreshMinInterval; concurrent - ## callers outside the throttle window coalesce onto a single in-flight refresh. + ## callers outside the throttle window coalesce onto a single in-flight + ## refresh, awaited via `join` so cancelling one caller never cancels it. if Moment.now() - g.lastRootsRefreshMoment < RootsRefreshMinInterval: return if not g.rootsRefreshInFlightFut.isNil() and not g.rootsRefreshInFlightFut.finished(): - await g.rootsRefreshInFlightFut + await g.rootsRefreshInFlightFut.join() return proc doRefresh(): Future[void] {.async.} = @@ -251,7 +252,7 @@ proc refreshRoots(g: OnchainGroupManager): Future[void] {.async.} = g.lastRootsRefreshMoment = Moment.now() g.rootsRefreshInFlightFut = doRefresh() - await g.rootsRefreshInFlightFut + await g.rootsRefreshInFlightFut.join() method validateRoot*(g: OnchainGroupManager, root: MerkleNode): Future[bool] {.async.} = if g.indexOfRoot(root) >= 0: @@ -260,45 +261,73 @@ method validateRoot*(g: OnchainGroupManager, root: MerkleNode): Future[bool] {.a await g.refreshRoots() return g.indexOfRoot(root) >= 0 +# Bounds refetches when invalidations keep landing mid-fetch, so a rejection +# storm cannot pin the refresh loop to the eth client. +const MerkleProofRefetchMaxAttempts = 3 + proc ensureFreshMerkleProofPath*( g: OnchainGroupManager -): Future[Result[void, string]] {.async.} = - ## Keeps `merkleProofCache` fresh independently of the validRoots window - ## used by the receive path. Refetches the path whenever the throttle - ## (`PathCheckMinInterval`) expires; trusts the cached path otherwise. - ## Guards against a missing membership index because `fetchMerkleProofElements` - ## unwraps it. +): Future[Result[seq[byte], string]] {.async.} = + ## Returns the merkle proof path, refetching from chain when the cache is + ## empty; suspected-stale callers invalidate first. Concurrent callers + ## coalesce onto one refetch, awaited via `join` so cancelling one caller + ## never cancels the shared refetch. Callers use the returned path, not the + ## cache field, so a concurrent invalidate cannot fail an in-flight proof-gen. if g.membershipIndex.isNone(): return err("membership index is not set") - if g.merkleProofCache.len > 0 and - Moment.now() - g.lastMerklePathCheckMoment < PathCheckMinInterval: - return ok() + if g.merkleProofCache.len > 0: + return ok(g.merkleProofCache) - if not g.proofPathRefreshInFlightFut.isNil() and - not g.proofPathRefreshInFlightFut.finished(): - await g.proofPathRefreshInFlightFut - if g.merkleProofCache.len > 0: - return ok() + proc doRefresh(): Future[seq[byte]] {.async.} = + var pathBytes: seq[byte] + for _ in 0 ..< MerkleProofRefetchMaxAttempts: + let generation = g.merkleProofCacheGeneration + pathBytes = (await g.fetchMerkleProofElements()).valueOr: + error "Failed to refresh merkle proof path", error = error + return @[] + if g.merkleProofCacheGeneration == generation: + g.merkleProofCache = pathBytes + break + # An invalidate raced this fetch, so the fetched path may predate the + # change that triggered it: fetch again. If attempts run out the last + # path is returned uncached and the next publish refetches. + if pathBytes.len > 0: + # Best-effort metric refresh - if there's a failure, it will update with the next root change + discard await g.updateMemberCount() + return pathBytes + + if g.proofPathRefreshInFlightFut.isNil() or g.proofPathRefreshInFlightFut.finished(): + g.proofPathRefreshInFlightFut = doRefresh() + + let refreshFut = g.proofPathRefreshInFlightFut + await refreshFut.join() + if not refreshFut.completed(): return err("merkle proof path refresh failed") - var fetchOk = false - proc doRefresh(): Future[void] {.async.} = - let pathBytes = (await g.fetchMerkleProofElements()).valueOr: - error "Failed to refresh merkle proof path", error = error - return - g.merkleProofCache = pathBytes - g.lastMerklePathCheckMoment = Moment.now() - fetchOk = true - # Best-effort metric refresh - if there's a failure, it will update with the next root change - discard await g.updateMemberCount() - - g.proofPathRefreshInFlightFut = doRefresh() - await g.proofPathRefreshInFlightFut - - if not fetchOk: + let pathBytes = refreshFut.read() + if pathBytes.len == 0: return err("merkle proof path refresh failed") - return ok() + return ok(pathBytes) + +method invalidateMerkleProofCache*(g: OnchainGroupManager) {.gcsafe, raises: [].} = + ## Empties the cache so the next `ensureFreshMerkleProofPath` refetches. The + ## generation bump makes a refetch already in flight fetch again rather than + ## serve a path that may predate this invalidate. + g.merkleProofCache = @[] + g.merkleProofCacheGeneration.inc() + +method scheduleMerkleProofRefresh*(g: OnchainGroupManager) {.gcsafe, raises: [].} = + ## Invalidates the cache and spawns a detached refetch; a failed refetch is + ## logged and repaired by the next `ensureFreshMerkleProofPath`. + g.invalidateMerkleProofCache() + + proc refresh() {.async.} = + let res = await g.ensureFreshMerkleProofPath() + if res.isErr(): + warn "merkle proof refresh failed", error = res.error + + asyncSpawn refresh() method register*( g: OnchainGroupManager, rateCommitment: RateCommitment @@ -485,9 +514,9 @@ method generateProof*( return err("user message limit is not set") debug "Generating RLN proof" - ?(await g.ensureFreshMerkleProofPath()) + let merkleProofPath = ?(await g.ensureFreshMerkleProofPath()) - if (g.merkleProofCache.len mod 32) != 0: + if (merkleProofPath.len mod 32) != 0: return err("Invalid merkle proof cache length") let identity_secret = seqToField(g.idCredentials.get().idSecretHash) @@ -496,8 +525,8 @@ method generateProof*( var path_elements = newSeq[byte](0) let identity_path_index = uint64ToIndex(g.membershipIndex.get(), 20) - for i in 0 ..< g.merkleProofCache.len div 32: - let chunk = g.merkleProofCache[i * 32 .. (i + 1) * 32 - 1] + for i in 0 ..< merkleProofPath.len div 32: + let chunk = merkleProofPath[i * 32 .. (i + 1) * 32 - 1] path_elements.add(chunk.reversed()) let xCfr = hashToFieldLe(data).valueOr: diff --git a/logos_delivery/waku/rln/proof.nim b/logos_delivery/waku/rln/proof.nim index fbbe2292e..8a29eb67b 100644 --- a/logos_delivery/waku/rln/proof.nim +++ b/logos_delivery/waku/rln/proof.nim @@ -64,9 +64,43 @@ proc generateRLNProof*( return err("could not generate rln-v2 proof: " & $error) return ok(proof.encode().buffer) +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: + return err("failed to generate RLN proof: " & $error) + + let rlnProof = RateLimitProof.init(proofBytes).valueOr: + return err("could not decode proof for root check: " & $error) + + if await rln.groupManager.validateRoot(rlnProof.merkleRoot): + return ok(proofBytes) + + info "RLN: stale merkle root detected; refreshing merkle path and regenerating proof" + rln.groupManager.invalidateMerkleProofCache() + return await rln.generateRLNProof(input, senderEpochTime) + +proc attachRLNProof*( + r: Rln, message: WakuMessage +): Future[Result[WakuMessage, string]] {.async.} = + ## Returns the message with a freshly generated RLN proof, replacing any + ## existing one and drawing a new message id. Retry paths suspecting a stale + ## path should call `invalidateMerkleProofCache` first. + var msgWithProof = message + msgWithProof.proof = ( + await r.generateRLNProof(message.toRLNSignal(), float64(getTime().toUnix())) + ).valueOr: + return err("error in attachRLNProof: " & error) + return ok(msgWithProof) + proc checkAndGenerateRLNProof*( rln: Option[Rln], message: WakuMessage ): Future[Result[WakuMessage, string]] {.async.} = + ## Returns the message with an attached RLN proof, or unchanged when it + ## already carries a proof or RLN is not configured. if message.proof.len > 0: return ok(message) @@ -74,12 +108,4 @@ proc checkAndGenerateRLNProof*( notice "Publishing message without RLN proof" return ok(message) - let - time = getTime().toUnix() - senderEpochTime = float64(time) - var msgWithProof = message - msgWithProof.proof = ( - await rln.get().generateRLNProof(msgWithProof.toRLNSignal(), senderEpochTime) - ).valueOr: - return err("error in checkAndGenerateRLNProof: " & $error) - return ok(msgWithProof) + return await attachRLNProof(rln.get(), message) diff --git a/logos_delivery/waku/rln/rln.nim b/logos_delivery/waku/rln/rln.nim index ec20cf0c8..fa952822b 100644 --- a/logos_delivery/waku/rln/rln.nim +++ b/logos_delivery/waku/rln/rln.nim @@ -233,10 +233,11 @@ proc mount( proc( msg: WakuMessage, senderEpochTime: float64 ): Future[Result[RequestGenerateRlnProof, string]] {.async.} = - let proof = (await rln.generateRLNProof(msg.toRLNSignal(), senderEpochTime)).valueOr: + let proofBytes = ( + await rln.generateRLNProofWithRootRefresh(msg.toRLNSignal(), senderEpochTime) + ).valueOr: return err("Could not create RLN proof: " & error) - - return ok(RequestGenerateRlnProof(proof: proof)), + return ok(RequestGenerateRlnProof(proof: proofBytes)), ).isOkOr: return err("Proof generator provider cannot be set: " & $error) diff --git a/tests/node/test_wakunode_legacy_lightpush.nim b/tests/node/test_wakunode_legacy_lightpush.nim index 9301e9e01..aec37e18c 100644 --- a/tests/node/test_wakunode_legacy_lightpush.nim +++ b/tests/node/test_wakunode_legacy_lightpush.nim @@ -1,7 +1,7 @@ {.used.} import - std/[options, tempfiles, net, osproc], + std/[options, tempfiles, net, osproc, strutils], testutils/unittests, chronos, std/strformat, @@ -16,6 +16,7 @@ import waku_lightpush_legacy/common, waku_lightpush_legacy/protocol_metrics, rln, + rln/constants, ], ../testlib/[wakucore, wakunode, testasync, futures, testutils], ../resources/payloads, @@ -120,7 +121,10 @@ suite "RLN Proofs as a Lightpush Service": # mount rln-relay # match prod epoch window to reduce test flake wakuRlnConfig = getWakuRlnConfig( - manager = manager, index = MembershipIndex(1), epochSizeSec = 600 + manager = manager, + userMessageLimit = 20, + index = MembershipIndex(1), + epochSizeSec = 600, ) await allFutures(server.start(), client.start()) @@ -180,6 +184,95 @@ suite "RLN Proofs as a Lightpush Service": assert publishResponse.isErr(), "We expect an error response" check publishResponse.error == protocol_metrics.notPublishedAnyPeer + # 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 + # pushHandler. Swapping in a stub pushHandler lets each test control what + # attempt N sees. + + asyncTest "retry fires on RlnValidatorErrorMsg substring and second attempt succeeds": + 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() + server.wakuLegacyLightPush.pushHandler = stub + + let response = await server.legacyLightpushPublish( + some(pubsubTopic), message, server.peerInfo.toRemotePeerInfo() + ) + + check: + callCount == 2 + response.isOk() + + asyncTest "no retry when error does not contain RlnValidatorErrorMsg": + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult[void]] {.async.} = + inc callCount + return err("unrelated failure") + server.wakuLegacyLightPush.pushHandler = stub + + let response = await server.legacyLightpushPublish( + some(pubsubTopic), message, server.peerInfo.toRemotePeerInfo() + ) + + check: + callCount == 1 + 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 + # teardown so server.stop() sees the same object graph it was + # constructed with. + let savedRln = server.rln + server.rln = nil + + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult[void]] {.async.} = + inc callCount + return err(RlnValidatorErrorMsg & ": simulated") + server.wakuLegacyLightPush.pushHandler = stub + + let response = await server.legacyLightpushPublish( + some(pubsubTopic), message, server.peerInfo.toRemotePeerInfo() + ) + + server.rln = savedRln + + check: + callCount == 1 + response.isErr() + suite "Waku Legacy Lightpush message delivery": asyncTest "Legacy lightpush message flow succeed": ## Setup diff --git a/tests/node/test_wakunode_lightpush.nim b/tests/node/test_wakunode_lightpush.nim index 6e87ab237..f13cbcaab 100644 --- a/tests/node/test_wakunode_lightpush.nim +++ b/tests/node/test_wakunode_lightpush.nim @@ -1,7 +1,7 @@ {.used.} import - std/[options, tempfiles, osproc], + std/[options, tempfiles, osproc, strutils], testutils/unittests, chronos, std/strformat, @@ -117,7 +117,10 @@ suite "RLN Proofs as a Lightpush Service": # mount rln-relay # match prod epoch window to reduce test flake wakuRlnConfig = getWakuRlnConfig( - manager = manager, index = MembershipIndex(1), epochSizeSec = 600 + manager = manager, + userMessageLimit = 20, + index = MembershipIndex(1), + epochSizeSec = 600, ) await allFutures(server.start(), client.start()) @@ -177,6 +180,182 @@ suite "RLN Proofs as a Lightpush Service": assert publishResponse.isErr(), "We expect an error response" check publishResponse.error.code == LightPushErrorCode.NO_PEERS_TO_RELAY + asyncTest "invalidate + regenerate refetches merkle path and rebuilds proof": + # Exercises the primitive pair that lightpushPublish leans on after a + # 420 (INVALID_MESSAGE) or 504 (OUT_OF_RLN_PROOF) rejection: calling + # invalidateMerkleProofCache empties the cached path so the next + # proof-gen refetches from chain, and attachRLNProof rebuilds the proof + # even though the message already carries one. + let firstMsg = (await checkAndGenerateRLNProof(some(server.rln), message)).get() + check firstMsg.proof.len > 0 + + # Corrupt the cache to model a stale/invalid witness — the same state a + # 420/504 rejection would leave us in. + let manager = cast[OnchainGroupManager](server.rln.groupManager) + let goodCache = manager.merkleProofCache + manager.merkleProofCache = newSeq[byte](goodCache.len) + check manager.merkleProofCache != goodCache + + # Retry path: invalidate the cache so the next proof-gen refetches from + # chain, then regenerate the proof. + manager.invalidateMerkleProofCache() + let secondMsg = (await attachRLNProof(server.rln, firstMsg)).get() + + check: + secondMsg.proof.len > 0 + # Regenerated, not passed through — Groth16 proofs carry random + # blinding, so a fresh call produces different bytes. + secondMsg.proof != firstMsg.proof + # Cache was refetched from chain, overwriting the corruption. + manager.merkleProofCache == goodCache + + # The tests below drive `server.lightpushPublish(...)` directly against the + # server node. Because `server.wakuLightPush` is mounted (and no lightpush + # client is), lightpushPublish takes the self-request path — it still runs + # the full client-side flow (proof gen, RLN-rejection classification), but + # the request lands in the local pushHandler instead of going over the + # wire. Swapping in a stub pushHandler lets each test control exactly what + # each attempt sees. + + asyncTest "RLN-related 420 surfaces as 504 and schedules a merkle proof refresh": + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult] {.async.} = + inc callCount + return + lighpushErrorResult(LightPushErrorCode.INVALID_MESSAGE, RlnValidatorErrorMsg) + server.wakuLightPush.pushHandler = stub + + let manager = cast[OnchainGroupManager](server.rln.groupManager) + let goodCache = manager.merkleProofCache + check goodCache.len > 0 + + let response = await server.lightpushPublish(some(pubsubTopic), message) + + check: + # The rejection is surfaced immediately — no internal republish — and + # normalized to 504 so callers can key their retry on it. + callCount == 1 + response.isErr() + response.error.code == LightPushErrorCode.OUT_OF_RLN_PROOF + response.error.desc.get("").contains(RlnProofRefreshScheduledMsg) + + # The refresh runs detached from the publish call: the cache was + # dropped and a refetch is in flight. Once it settles the path is + # fresh again. + let inFlight = manager.proofPathRefreshInFlightFut + if not inFlight.isNil(): + await inFlight.join() + check manager.merkleProofCache == goodCache + + asyncTest "504 (OUT_OF_RLN_PROOF) from the service passes through without republish": + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult] {.async.} = + inc callCount + return lighpushErrorResult( + LightPushErrorCode.OUT_OF_RLN_PROOF, "simulated out-of-proof" + ) + server.wakuLightPush.pushHandler = stub + + let response = await server.lightpushPublish(some(pubsubTopic), message) + + check: + callCount == 1 + response.isErr() + response.error.code == LightPushErrorCode.OUT_OF_RLN_PROOF + + asyncTest "caller retry after 504 lands on the refreshed merkle path": + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult] {.async.} = + inc callCount + if callCount == 1: + return lighpushErrorResult( + LightPushErrorCode.INVALID_MESSAGE, RlnValidatorErrorMsg + ) + return lightpushSuccessResult(1) + server.wakuLightPush.pushHandler = stub + + let firstResponse = await server.lightpushPublish(some(pubsubTopic), message) + + check: + firstResponse.isErr() + firstResponse.error.code == LightPushErrorCode.OUT_OF_RLN_PROOF + + # Model the caller-level retry (send service next round, REST handler): + # a second publish regenerates the proof against the refreshed cache. + let secondResponse = await server.lightpushPublish(some(pubsubTopic), message) + + check: + callCount == 2 + secondResponse.isOk() + secondResponse.get() == 1 + + asyncTest "420 without the RLN error text passes through unchanged": + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult] {.async.} = + inc callCount + return + lighpushErrorResult(LightPushErrorCode.INVALID_MESSAGE, "unrelated rejection") + server.wakuLightPush.pushHandler = stub + + let response = await server.lightpushPublish(some(pubsubTopic), message) + + check: + callCount == 1 + response.isErr() + response.error.code == LightPushErrorCode.INVALID_MESSAGE + + asyncTest "error codes outside 420/504 pass through unchanged": + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult] {.async.} = + inc callCount + return lighpushErrorResult( + LightPushErrorCode.INTERNAL_SERVER_ERROR, "unrelated failure" + ) + server.wakuLightPush.pushHandler = stub + + let response = await server.lightpushPublish(some(pubsubTopic), message) + + check: + callCount == 1 + response.isErr() + response.error.code == LightPushErrorCode.INTERNAL_SERVER_ERROR + + asyncTest "rejection passes through unchanged when node.rln is nil": + # Detach RLN so the RLN-rejection branch short-circuits on rln.isNone() + # even for a 420. Restore before teardown so server.stop() sees the same + # object graph it was constructed with. + let savedRln = server.rln + server.rln = nil + + var callCount = 0 + let stub: PushMessageHandler = proc( + pubsubTopic: PubsubTopic, message: WakuMessage + ): Future[WakuLightPushResult] {.async.} = + inc callCount + return lighpushErrorResult( + LightPushErrorCode.INVALID_MESSAGE, "simulated stale merkle path" + ) + server.wakuLightPush.pushHandler = stub + + let response = await server.lightpushPublish(some(pubsubTopic), message) + + server.rln = savedRln + + check: + callCount == 1 + response.isErr() + response.error.code == LightPushErrorCode.INVALID_MESSAGE + suite "Waku Lightpush message delivery": asyncTest "lightpush message flow succeed": ## Setup diff --git a/tests/waku_rln_relay/test_rln_group_manager_onchain.nim b/tests/waku_rln_relay/test_rln_group_manager_onchain.nim index 491da4274..bbd97217f 100644 --- a/tests/waku_rln_relay/test_rln_group_manager_onchain.nim +++ b/tests/waku_rln_relay/test_rln_group_manager_onchain.nim @@ -303,9 +303,6 @@ suite "Onchain group manager": # chunk[0] becomes the MSB after reversal in group_manager; must be < 0x30 for i in 0 ..< 20: manager.merkleProofCache[i * 32] = 0 - # Pin the freshness throttle so ensureFreshMerkleProofPath does NOT refetch - # and overwrite the intentionally-corrupted cache we just planted. - manager.lastMerklePathCheckMoment = Moment.now() let messageBytes = "Hello".toBytes() @@ -435,7 +432,7 @@ suite "Onchain group manager": # replaced by a competing refresh. manager.rootsRefreshInFlightFut == inFlight - test "generateProof: fast-paths without refresh inside throttle window": + test "validateRoot: cancelling one caller does not cancel the shared roots refresh": (waitFor manager.init()).isOkOr: raiseAssert $error @@ -443,11 +440,47 @@ suite "Onchain group manager": (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: assert false, "register failed: " & error - # Prime cache and pin the throttle so the publish-path freshness check - # short-circuits on the cached value. + manager.lastRootsRefreshMoment = default(Moment) + manager.rootsRefreshInFlightFut = nil + + var badRoot: MerkleNode + badRoot[0] = 0x66 + + let f1 = manager.validateRoot(badRoot) + let inFlight = manager.rootsRefreshInFlightFut + let f2 = manager.validateRoot(badRoot) + + check: + inFlight != nil + not inFlight.finished() + + # Cancel the initiating caller; the shared refresh must keep running for + # the coalesced one. + waitFor f1.cancelAndWait() + + check: + f1.cancelled() + not inFlight.cancelled() + + discard waitFor f2 + + check: + inFlight.completed() + manager.rootsRefreshInFlightFut == inFlight + + test "generateProof: fast-paths without refresh when cache is populated": + (waitFor manager.init()).isOkOr: + raiseAssert $error + + let credentials = generateCredentials() + (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: + assert false, "register failed: " & error + + # Prime cache. A non-empty cache short-circuits ensureFreshMerkleProofPath; + # the retry path invalidates the cache via invalidateMerkleProofCache before + # calling proof-gen again. manager.merkleProofCache = (waitFor manager.fetchMerkleProofElements()).valueOr: raiseAssert "failed to fetch initial path: " & error - manager.lastMerklePathCheckMoment = Moment.now() manager.proofPathRefreshInFlightFut = nil let primedCache = manager.merkleProofCache @@ -498,7 +531,7 @@ suite "Onchain group manager": res.isErr() res.error == "membership index is not set" - test "ensureFreshMerkleProofPath: refetches when throttle window has expired": + test "invalidateMerkleProofCache + ensureFreshMerkleProofPath refetches": (waitFor manager.init()).isOkOr: raiseAssert $error @@ -506,22 +539,21 @@ suite "Onchain group manager": (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: assert false, "register failed: " & error - # Prime cache with a non-empty value and an old throttle timestamp, so - # the cache fast-path does NOT trigger and we exercise the refetch branch. + # Prime cache with a non-empty value, then take the retry path: + # invalidate empties the cache so ensureFreshMerkleProofPath refetches. manager.merkleProofCache = (waitFor manager.fetchMerkleProofElements()).valueOr: raiseAssert "failed to prime path: " & error - manager.lastMerklePathCheckMoment = Moment.now() - PathCheckMinInterval - 1.seconds manager.proofPathRefreshInFlightFut = nil - let preCheckTs = manager.lastMerklePathCheckMoment + manager.invalidateMerkleProofCache() + check manager.merkleProofCache.len == 0 + let res = waitFor manager.ensureFreshMerkleProofPath() check: res.isOk() manager.merkleProofCache.len > 0 manager.proofPathRefreshInFlightFut != nil - # lastMerklePathCheckMoment was bumped to "now" by the refetch. - manager.lastMerklePathCheckMoment > preCheckTs test "ensureFreshMerkleProofPath: refresh bumps the member-count metric": (waitFor manager.init()).isOkOr: @@ -554,12 +586,10 @@ suite "Onchain group manager": (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: assert false, "register failed: " & error - # Empty cache + epoch-zero check timestamp guarantees the first caller - # will fall through to fetchMerkleProofElements; the followers should - # observe the resulting in-flight future and await it rather than start - # their own refresh. + # Empty cache guarantees the first caller will fall through to + # fetchMerkleProofElements; the followers should observe the resulting + # in-flight future and await it rather than start their own refresh. manager.merkleProofCache = @[] - manager.lastMerklePathCheckMoment = default(Moment) manager.proofPathRefreshInFlightFut = nil let f1 = manager.ensureFreshMerkleProofPath() @@ -589,6 +619,130 @@ suite "Onchain group manager": manager.proofPathRefreshInFlightFut == inFlight manager.merkleProofCache.len > 0 + test "ensureFreshMerkleProofPath: cancelling one caller does not cancel the shared refetch": + (waitFor manager.init()).isOkOr: + raiseAssert $error + + let credentials = generateCredentials() + (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: + assert false, "register failed: " & error + + manager.merkleProofCache = @[] + manager.proofPathRefreshInFlightFut = nil + + let f1 = manager.ensureFreshMerkleProofPath() + let inFlight = manager.proofPathRefreshInFlightFut + let f2 = manager.ensureFreshMerkleProofPath() + + check: + inFlight != nil + not inFlight.finished() + + # Cancel the initiating caller — models a publish retry whose refresh + # timeout fired while the refetch was still in flight. + waitFor f1.cancelAndWait() + + check: + f1.cancelled() + # The shared refetch survives the caller's cancellation. + not inFlight.cancelled() + + # The coalesced caller still completes and the cache gets populated. + let r2 = waitFor f2 + check: + r2.isOk() + manager.merkleProofCache.len > 0 + manager.proofPathRefreshInFlightFut == inFlight + + test "scheduleMerkleProofRefresh: drops the cache and refetches in the background": + (waitFor manager.init()).isOkOr: + raiseAssert $error + + let credentials = generateCredentials() + (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: + assert false, "register failed: " & error + + # Prime the cache, then corrupt it to model the stale path a publish + # rejection points at. + let goodCache = (waitFor manager.fetchMerkleProofElements()).valueOr: + raiseAssert "failed to prime path: " & error + manager.merkleProofCache = newSeq[byte](goodCache.len) + manager.proofPathRefreshInFlightFut = nil + + manager.scheduleMerkleProofRefresh() + + # The call returns without waiting; the refetch is already in flight. + let inFlight = manager.proofPathRefreshInFlightFut + check inFlight != nil + + waitFor inFlight.join() + check manager.merkleProofCache == goodCache + + test "ensureFreshMerkleProofPath: invalidate during the refetch does not fail callers": + (waitFor manager.init()).isOkOr: + raiseAssert $error + + let credentials = generateCredentials() + (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: + assert false, "register failed: " & error + + manager.merkleProofCache = @[] + manager.proofPathRefreshInFlightFut = nil + + # Both callers coalesce onto a refetch suspended on the eth call. + let f1 = manager.ensureFreshMerkleProofPath() + let inFlight = manager.proofPathRefreshInFlightFut + let f2 = manager.ensureFreshMerkleProofPath() + + check: + inFlight != nil + not inFlight.finished() + + # A publish rejection lands mid-fetch. The invalidate must neither fail + # the coalesced callers nor be swallowed by the fetch it raced: the path + # they receive is refetched after the invalidate. + manager.invalidateMerkleProofCache() + check manager.merkleProofCache.len == 0 + + let r1 = waitFor f1 + let r2 = waitFor f2 + + check: + r1.isOk() + r1.get().len > 0 + r2.isOk() + r2.get() == r1.get() + # The post-invalidate refetch repopulated the cache. + manager.merkleProofCache == r1.get() + + test "generateProof: succeeds when the cache is invalidated mid-refetch": + (waitFor manager.init()).isOkOr: + raiseAssert $error + + let credentials = generateCredentials() + (waitFor manager.register(credentials, UserMessageLimit(20))).isOkOr: + assert false, "register failed: " & error + + manager.merkleProofCache = @[] + manager.proofPathRefreshInFlightFut = nil + + # generateProof runs down to the refetch's suspended eth call. + let proofFut = manager.generateProof( + data = "hello".toBytes(), epoch = default(Epoch), messageId = MessageId(1) + ) + + check: + manager.proofPathRefreshInFlightFut != nil + not manager.proofPathRefreshInFlightFut.finished() + + # A concurrent publish rejection empties the cache while proof-gen waits + # on the refetch. + manager.invalidateMerkleProofCache() + + let proofRes = waitFor proofFut + check: + proofRes.isOk() + test "verifyProof: should verify valid proof": let credentials = generateCredentials() (waitFor manager.init()).isOkOr: @@ -657,9 +811,6 @@ suite "Onchain group manager": # chunk[0] becomes the MSB after reversal in group_manager; must be < 0x30 for i in 0 ..< 20: manager.merkleProofCache[i * 32] = 0 - # Pin the freshness throttle so ensureFreshMerkleProofPath does NOT refetch - # and overwrite the intentionally-corrupted cache we just planted. - manager.lastMerklePathCheckMoment = Moment.now() let epoch = default(Epoch) info "epoch in bytes", epochHex = epoch.inHex() diff --git a/tests/waku_rln_relay/test_wakunode_rln_relay.nim b/tests/waku_rln_relay/test_wakunode_rln_relay.nim index 7954317e7..4c02b4cbd 100644 --- a/tests/waku_rln_relay/test_wakunode_rln_relay.nim +++ b/tests/waku_rln_relay/test_wakunode_rln_relay.nim @@ -11,7 +11,8 @@ import brokers/broker_context import - logos_delivery/waku/[waku_core, waku_node, rln], + logos_delivery/waku/[waku_core, waku_node, rln, rln/protocol_types], + logos_delivery/waku/requests/rln_requests, ../testlib/[wakucore, futures, wakunode, testutils], ./utils_onchain, ./rln/waku_rln_relay_utils @@ -751,3 +752,55 @@ procSuite "WakuNode - RLN relay": # Cleanup waitFor allFutures(node1.stop(), node2.stop()) + + asyncTest "broker proof provider retries with force-refresh when initial proof has stale root": + ## Exercises the reactive mechanism added to RequestGenerateRlnProof.setProvider + ## in rln.nim: when the cached Merkle proof path produces a proof with a root + ## that validateRoot rejects, the provider must force-refresh the path and + ## return a proof whose root is in the valid-roots window. + lockNewGlobalBrokerContext: + let nodeKey = generateSecp256k1Key() + let node = newTestWakuNode(nodeKey) + (await node.mountRelay()).isOkOr: + assert false, "Failed to mount relay" + + let wakuRlnConfig = getWakuRlnConfig( + manager = manager, + index = MembershipIndex(1), + epochSizeSec = 600, + userMessageLimit = 20, + ) + await node.setRlnValidator(wakuRlnConfig) + await node.start() + + let rlnManager = cast[OnchainGroupManager](node.rln.groupManager) + let idCredentials = generateCredentials() + (waitFor rlnManager.register(idCredentials, UserMessageLimit(20))).isOkOr: + assert false, "Failed to register: " & error + + let rootUpdated = waitFor rlnManager.updateRoots() + info "Updated root", rootUpdated + + let proofRes = waitFor rlnManager.fetchMerkleProofElements() + assert proofRes.isOk(), "failed to fetch merkle proof: " & proofRes.error + let goodCache = proofRes.get() + rlnManager.merkleProofCache = goodCache + + # Corrupt the cache so the first generateRLNProof inside the provider + # produces a proof with a Merkle root that is not in the valid-roots window. + # The provider must detect this via validateRoot, force-refresh, and retry. + rlnManager.merkleProofCache = newSeq[byte](goodCache.len) + + let msg = fakeWakuMessage() + let proofResult = + await RequestGenerateRlnProof.request(node.rln.brokerCtx, msg, epochTime()) + + check proofResult.isOk() + # The force-refresh inside the provider restored the correct path + check rlnManager.merkleProofCache == goodCache + # The returned proof carries a Merkle root that is in the valid-roots window + let rlnProof = RateLimitProof.init(proofResult.get().proof).get() + let rootValid = await node.rln.groupManager.validateRoot(rlnProof.merkleRoot) + check rootValid + + await node.stop() diff --git a/tests/wakunode_rest/test_rest_relay.nim b/tests/wakunode_rest/test_rest_relay.nim index ccf59b4fd..b59dc463d 100644 --- a/tests/wakunode_rest/test_rest_relay.nim +++ b/tests/wakunode_rest/test_rest_relay.nim @@ -1,7 +1,7 @@ {.used.} import - std/[sequtils, strformat, tempfiles, osproc, options], + std/[sequtils, strformat, strutils, tempfiles, osproc, options], stew/byteutils, testutils/unittests, presto, @@ -793,3 +793,199 @@ suite "Waku v2 Rest API - Relay": await restServer.stop() await restServer.closeWait() await node.stop() + + asyncTest "Stale RLN proof returns 503 and schedules a refresh - POST /relay/v1/messages/{topic}": + ## When the cached Merkle proof path is stale the handler generates a proof + ## whose root the local RLN validator rejects. The handler must detect the + ## RlnValidatorErrorMsg, schedule a background merkle proof refresh, and + ## fail early with 503 + RlnProofRefreshScheduledMsg. A client retry then + ## succeeds against the refreshed path. + let node = testWakuNode() + (await node.mountRelay()).isOkOr: + assert false, "Failed to mount relay" + let wakuRlnConfig = getWakuRlnConfig( + manager = manager, + index = MembershipIndex(1), + epochSizeSec = 600, + userMessageLimit = 20, + ) + await node.setRlnValidator(wakuRlnConfig) + await node.start() + + let manager = cast[OnchainGroupManager](node.rln.groupManager) + let idCredentials = generateCredentials() + (waitFor manager.register(idCredentials, UserMessageLimit(20))).isOkOr: + assert false, "Failed to register: " & getCurrentExceptionMsg() + + let rootUpdated = waitFor manager.updateRoots() + info "Updated root", rootUpdated + + let proofRes = waitFor manager.fetchMerkleProofElements() + assert proofRes.isOk(), "failed to fetch merkle proof: " & proofRes.error + let goodCache = proofRes.get() + manager.merkleProofCache = goodCache + + # Corrupt the cache with zeros so the first generateRLNProof call produces a + # proof with a Merkle root that is not in the valid-roots window. + # validateMessage will return RlnValidatorErrorMsg. + manager.merkleProofCache = newSeq[byte](goodCache.len) + + var restPort = Port(0) + let restAddress = parseIpAddress("0.0.0.0") + let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet() + restPort = restServer.httpServer.address.port + let cache = MessageCache.init() + installRelayApiHandlers(restServer.router, node, cache) + restServer.start() + let client = newRestHttpClient(initTAddress(restAddress, restPort)) + + let simpleHandler = proc( + topic: PubsubTopic, msg: WakuMessage + ): Future[void] {.async, gcsafe.} = + await sleepAsync(0.milliseconds) + + node.subscribe((kind: PubsubSub, topic: DefaultPubsubTopic), simpleHandler).isOkOr: + assert false, "Failed to subscribe to pubsub topic" + + let response = await client.relayPostMessagesV1( + DefaultPubsubTopic, + RelayWakuMessage( + payload: base64.encode("TEST-PAYLOAD"), + contentTopic: some(DefaultContentTopic), + timestamp: some(now()), + ), + ) + + # The handler fails early with the retry signal; the refresh runs detached. + check: + response.status == 503 + $response.contentType == $MIMETYPE_TEXT + response.data.contains(RlnProofRefreshScheduledMsg) + + let inFlight = manager.proofPathRefreshInFlightFut + if not inFlight.isNil(): + await inFlight.join() + check manager.merkleProofCache == goodCache # refresh restored the correct path + + # A client retry now succeeds against the refreshed path. + let retryResponse = await client.relayPostMessagesV1( + DefaultPubsubTopic, + RelayWakuMessage( + payload: base64.encode("TEST-PAYLOAD"), + contentTopic: some(DefaultContentTopic), + timestamp: some(now()), + ), + ) + + check: + retryResponse.status == 200 + retryResponse.data == "OK" + + await restServer.stop() + await restServer.closeWait() + await node.stop() + + asyncTest "Stale RLN proof returns 503 and schedules a refresh - POST /relay/v1/auto/messages/{topic}": + ## Same fail-fast behavior as the static-sharding handler, exercised via + ## the auto-sharding endpoint. A relay-only mesh node is connected so that + ## node.publish() has a gossipsub peer and the client retry can return + ## success. + + # Relay-only mesh node — no RLN needed, just provides a gossipsub peer. + let meshNode = testWakuNode() + (await meshNode.mountRelay()).isOkOr: + assert false, "Failed to mount relay on mesh node" + require meshNode.mountAutoSharding(1, 8).isOk + await meshNode.start() + let meshHandler = proc( + topic: PubsubTopic, msg: WakuMessage + ): Future[void] {.async, gcsafe.} = + discard + meshNode.subscribe((kind: ContentSub, topic: DefaultContentTopic), meshHandler).isOkOr: + assert false, "Failed to subscribe mesh node" + + var node: WakuNode + lockNewGlobalBrokerContext: + node = testWakuNode() + (await node.mountRelay()).isOkOr: + assert false, "Failed to mount relay" + require node.mountAutoSharding(1, 8).isOk + + let wakuRlnConfig = getWakuRlnConfig( + manager = manager, + index = MembershipIndex(1), + epochSizeSec = 600, + userMessageLimit = 20, + ) + await node.setRlnValidator(wakuRlnConfig) + await node.start() + await node.connectToNodes(@[meshNode.peerInfo.toRemotePeerInfo()]) + + let manager = cast[OnchainGroupManager](node.rln.groupManager) + let idCredentials = generateCredentials() + (waitFor manager.register(idCredentials, UserMessageLimit(20))).isOkOr: + assert false, "Failed to register: " & getCurrentExceptionMsg() + + let rootUpdated = waitFor manager.updateRoots() + info "Updated root", rootUpdated + + let proofRes = waitFor manager.fetchMerkleProofElements() + assert proofRes.isOk(), "failed to fetch merkle proof: " & proofRes.error + let goodCache = proofRes.get() + manager.merkleProofCache = goodCache + + # Corrupt the cache to produce a proof with a bad Merkle root + manager.merkleProofCache = newSeq[byte](goodCache.len) + + var restPort = Port(0) + let restAddress = parseIpAddress("0.0.0.0") + let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet() + restPort = restServer.httpServer.address.port + let cache = MessageCache.init() + installRelayApiHandlers(restServer.router, node, cache) + restServer.start() + let client = newRestHttpClient(initTAddress(restAddress, restPort)) + + let simpleHandler = proc( + topic: PubsubTopic, msg: WakuMessage + ): Future[void] {.async, gcsafe.} = + await sleepAsync(0.milliseconds) + + node.subscribe((kind: ContentSub, topic: DefaultContentTopic), simpleHandler).isOkOr: + assert false, "Failed to subscribe to content topic" + + let response = await client.relayPostAutoMessagesV1( + RelayWakuMessage( + payload: base64.encode("TEST-PAYLOAD"), + contentTopic: some(DefaultContentTopic), + timestamp: some(now()), + ) + ) + + # The handler fails early with the retry signal; the refresh runs detached. + check: + response.status == 503 + $response.contentType == $MIMETYPE_TEXT + response.data.contains(RlnProofRefreshScheduledMsg) + + let inFlight = manager.proofPathRefreshInFlightFut + if not inFlight.isNil(): + await inFlight.join() + check manager.merkleProofCache == goodCache + + # A client retry now succeeds against the refreshed path. + let retryResponse = await client.relayPostAutoMessagesV1( + RelayWakuMessage( + payload: base64.encode("TEST-PAYLOAD"), + contentTopic: some(DefaultContentTopic), + timestamp: some(now()), + ) + ) + + check: + retryResponse.status == 200 + retryResponse.data == "OK" + + await restServer.stop() + await restServer.closeWait() + await allFutures(node.stop(), meshNode.stop())