diff --git a/logos_delivery/waku/node/waku_node/lightpush.nim b/logos_delivery/waku/node/waku_node/lightpush.nim index 1bb4ddcbb..d33bea34b 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, @@ -92,7 +92,7 @@ proc legacyLightpushPublish*( none(Rln) else: some(node.rln) - let msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: + var msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: return err("failed call checkAndGenerateRLNProof from lightpush: " & error) let internalPublish = proc( @@ -119,16 +119,52 @@ proc legacyLightpushPublish*( return await node.wakuLegacyLightPush.handleSelfLightPushRequest(pubsubTopic, message) try: + # Resolve the effective pubsub topic first (explicit param, else derive + # from autosharding), then run the send. We need a single resolved topic + # so the retry path below can reuse it without repeating this branch. + var pubsubForPublish: PubsubTopic if pubsubTopic.isSome(): - return await internalPublish(node, pubsubTopic.get(), msgWithProof, peer) + pubsubForPublish = pubsubTopic.get() + else: + if node.wakuAutoSharding.isNone(): + return err("Pubsub topic must be specified when static sharding is enabled") + let topicMap = + ?node.wakuAutoSharding.get().getShardsFromContentTopics(message.contentTopic) + var resolved = false + for pubsub, _ in topicMap.pairs: # There's only one pair anyway + pubsubForPublish = $pubsub + resolved = true + break + if not resolved: + # Preserve pre-existing behavior: an empty topicMap fell off the end + # of the loop and returned the default-initialized Result. + return - 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 internalPublish(node, pubsubForPublish, msgWithProof, peer) - for pubsub, _ in topicMap.pairs: # There's only one pair anyway - return await internalPublish(node, $pubsub, msgWithProof, peer) + # Legacy protocol has no status code taxonomy: the server collapses every + # failure into isSuccess=false + a free-text info string. The only stable + # substring we can safely branch on is "RLN validation failed" — the + # errorMessage registered for the RLN validator in waku_node/relay.nim. + # Match it and treat it as the legacy equivalent of a v3 420/504: force + # refresh the cached merkle path, regenerate the RLN proof, and retry the + # send exactly once. All other failure modes (decode error, rate limit, + # no peers) are surfaced to the caller unchanged because a fresh proof + # would not change their outcome. + if firstResult.isOk() or rln.isNone() or + not firstResult.error.contains("RLN validation failed"): + return firstResult + + info "legacy lightpush send rejected as RLN-invalid; " & + "refreshing merkle proof and retrying once" + msgWithProof = ( + await checkAndGenerateRLNProof(rln, msgWithProof, forceMerkleProofRefresh = true) + ).valueOr: + return err( + "failed call checkAndGenerateRLNProof from lightpush retry: " & error + ) + + return await internalPublish(node, pubsubForPublish, msgWithProof, peer) except CatchableError: return err(getCurrentExceptionMsg()) @@ -290,7 +326,26 @@ proc lightpushPublish*( none(Rln) else: some(node.rln) - let msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: + var msgWithProof = (await checkAndGenerateRLNProof(rln, message)).valueOr: + return lighpushErrorResult(LightPushErrorCode.OUT_OF_RLN_PROOF, error) + + let firstResult = + await lightpushPublishHandler(node, pubsubForPublish, msgWithProof, toPeer, mixify) + + # If message is rejected with error code 420 (INVALID_MESSAGE) or 504 + # (OUT_OF_RLN_PROOF) then cached merkle path is likely stale relative + # to the current on-chain group. Force-refresh it, regenerate the proof, + # and retry the send exactly once. + if firstResult.isOk() or rln.isNone() or + firstResult.error.code notin + [LightPushErrorCode.INVALID_MESSAGE, LightPushErrorCode.OUT_OF_RLN_PROOF]: + return firstResult + + info "lightpush send rejected; refreshing merkle proof and retrying once", + statusCode = $firstResult.error.code + msgWithProof = ( + await checkAndGenerateRLNProof(rln, msgWithProof, forceMerkleProofRefresh = true) + ).valueOr: return lighpushErrorResult(LightPushErrorCode.OUT_OF_RLN_PROOF, error) return diff --git a/logos_delivery/waku/rln/constants.nim b/logos_delivery/waku/rln/constants.nim index 22454b22b..961bbcce7 100644 --- a/logos_delivery/waku/rln/constants.nim +++ b/logos_delivery/waku/rln/constants.nim @@ -14,11 +14,6 @@ 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" 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..708e3252c 100644 --- a/logos_delivery/waku/rln/group_manager/group_manager_base.nim +++ b/logos_delivery/waku/rln/group_manager/group_manager_base.nim @@ -137,6 +137,7 @@ method generateProof*( epoch: Epoch, messageId: MessageId, rlnIdentifier = DefaultRlnIdentifier, + forceMerkleProofRefresh: bool = false, ): Future[GroupManagerResult[RateLimitProof]] {.base, async.} = ## Dummy implementation for generateProof return err("generateProof is not implemented") 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..3290e9b1a 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,7 +44,6 @@ type registrationHandler*: Option[RegistrationHandler] latestProcessedBlock*: BlockNumber merkleProofCache*: seq[byte] - lastMerklePathCheckMoment*: Moment proofPathRefreshInFlightFut*: Future[void] lastRootsRefreshMoment*: Moment rootsRefreshInFlightFut*: Future[void] @@ -261,18 +260,18 @@ method validateRoot*(g: OnchainGroupManager, root: MerkleNode): Future[bool] {.a return g.indexOfRoot(root) >= 0 proc ensureFreshMerkleProofPath*( - g: OnchainGroupManager + g: OnchainGroupManager, force: bool = false ): 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. + ## Keeps `merkleProofCache` fresh. Refetches only when the cache is empty + ## or when the caller explicitly forces a refresh — typically in response + ## to a lightpush 420 (INVALID_MESSAGE) or 504 (OUT_OF_RLN_PROOF) reply + ## that suggests the cached path is stale. Otherwise trusts the cache. ## Guards against a missing membership index because `fetchMerkleProofElements` ## unwraps it. if g.membershipIndex.isNone(): return err("membership index is not set") - if g.merkleProofCache.len > 0 and - Moment.now() - g.lastMerklePathCheckMoment < PathCheckMinInterval: + if g.merkleProofCache.len > 0 and not force: return ok() if not g.proofPathRefreshInFlightFut.isNil() and @@ -288,7 +287,6 @@ proc ensureFreshMerkleProofPath*( 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() @@ -474,6 +472,7 @@ method generateProof*( epoch: Epoch, messageId: MessageId, rlnIdentifier = DefaultRlnIdentifier, + forceMerkleProofRefresh: bool = false, ): Future[GroupManagerResult[RateLimitProof]] {.async.} = ## Generates an RLN proof using the cached Merkle proof and custom witness # Ensure identity credentials and membership index are set @@ -485,7 +484,7 @@ method generateProof*( return err("user message limit is not set") debug "Generating RLN proof" - ?(await g.ensureFreshMerkleProofPath()) + ?(await g.ensureFreshMerkleProofPath(force = forceMerkleProofRefresh)) if (g.merkleProofCache.len mod 32) != 0: return err("Invalid merkle proof cache length") diff --git a/logos_delivery/waku/rln/proof.nim b/logos_delivery/waku/rln/proof.nim index 4a0c332c1..c61a55a07 100644 --- a/logos_delivery/waku/rln/proof.nim +++ b/logos_delivery/waku/rln/proof.nim @@ -55,19 +55,29 @@ proc toRLNSignal*(wakumessage: WakuMessage): seq[byte] = return output proc generateRLNProof*( - rln: Rln, input: seq[byte], senderEpochTime: float64 + rln: Rln, + input: seq[byte], + senderEpochTime: float64, + forceMerkleProofRefresh: bool = false, ): Future[RlnResult[seq[byte]]] {.async.} = 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: + let proof = ( + await rln.groupManager.generateProof( + input, epoch, nonce, forceMerkleProofRefresh = forceMerkleProofRefresh + ) + ).valueOr: return err("could not generate rln-v2 proof: " & $error) return ok(proof.encode().buffer) proc checkAndGenerateRLNProof*( - rln: Option[Rln], message: WakuMessage + rln: Option[Rln], message: WakuMessage, forceMerkleProofRefresh: bool = false ): Future[Result[WakuMessage, string]] {.async.} = - if message.proof.len > 0: + # When forcing a refresh (e.g. retry after a lightpush 420/504 rejection) we + # deliberately regenerate the proof even if one is already attached, since + # the existing proof is presumed to have been rejected. + if message.proof.len > 0 and not forceMerkleProofRefresh: return ok(message) if rln.isNone(): @@ -79,7 +89,11 @@ proc checkAndGenerateRLNProof*( senderEpochTime = float64(time) var msgWithProof = message msgWithProof.proof = ( - await rln.get().generateRLNProof(msgWithProof.toRLNSignal(), senderEpochTime) + await rln.get().generateRLNProof( + msgWithProof.toRLNSignal(), + senderEpochTime, + forceMerkleProofRefresh = forceMerkleProofRefresh, + ) ).valueOr: return err("error in checkAndGenerateRLNProof: " & $error) return ok(msgWithProof) 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..a51efa353 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 "generateProof: fast-paths without refresh when cache is populated": (waitFor manager.init()).isOkOr: raiseAssert $error @@ -443,11 +440,10 @@ 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. + # Prime cache. With forceMerkleProofRefresh=false (the default), a non-empty + # cache short-circuits ensureFreshMerkleProofPath. 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 +494,7 @@ suite "Onchain group manager": res.isErr() res.error == "membership index is not set" - test "ensureFreshMerkleProofPath: refetches when throttle window has expired": + test "ensureFreshMerkleProofPath: refetches when caller forces a refresh": (waitFor manager.init()).isOkOr: raiseAssert $error @@ -506,22 +502,18 @@ 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. force=true must bypass the + # "cache is populated" fast-path and exercise the refetch branch. 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 - let res = waitFor manager.ensureFreshMerkleProofPath() + let res = waitFor manager.ensureFreshMerkleProofPath(force = true) 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 +546,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() @@ -657,9 +647,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()