refresh merkle proof cache reactively on lightpush rejection

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

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"
 

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")

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")

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)

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()