mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-08-03 18:33:15 +00:00
feat: attach and refresh RLN proofs in the send service (Stream B) (#4069)
* feat: attach RLN proofs at the SendService transmission stage
The relay send path published without an RLN proof: proof generation
lived client-side in (legacy)lightpushPublish, so messages dispatched
through SendService -> RelaySendProcessor reached the network unproven
and would be rejected by an RLN-enforcing relay.
Adds Waku.attachRlnProof in the waku/api publish surface and calls it
from SendService immediately after admission, in both send() and the
retry loop. Placement is load-bearing:
- After admit(), so a message rejected by the rate limiter never draws
a nonce.
- At transmission rather than API entry, because a proof binds to the
epoch current when the message goes out, and a task can be retried
for up to MaxTimeInCache after send() returns.
attachRlnProof is a no-op without RLN mounted (message passes through
unproven, as today) and short-circuits on a message that already
carries a proof, so retrying a task neither redraws a nonce nor
changes the bytes. It uses generateRLNProofWithRootRefresh rather than
the plain generator: a task can wait in the task cache while the group
root moves on chain, so the proof is validated against the
acceptable-root window and regenerated once against a refetched merkle
path if it went stale.
Proof-generation failure parks the task as NextRoundRetry rather than
failing it, matching the admission path: the dominant failure is
NonceLimitReached (RLN's own per-epoch budget exhausted), which the
service loop resolves as the epoch rolls over.
Adds tests/messaging/test_rln_proof_attach.nim covering the unmounted
pass-through, attach when mounted, and the idempotency contract that
the retry loop depends on.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat: handle RLN publish rejections in the send service retry loop
An RLN-invalid publish rejection now recovers through the send
service's existing retry loop instead of an inline retry in the kernel.
When a relay or lightpush publish is rejected as RLN-invalid, the
processor clears the message's stale proof, schedules a background
merkle-proof refresh, and parks the task as NextRoundRetry. The next
loop round re-admits the task and regenerates the proof against the
refreshed path. Clearing the proof is required: attachRlnProof
short-circuits on a message that already carries one, so without the
clear the task would resend the rejected proof until it ages out. The
relay processor previously failed such tasks outright, with no
recovery.
Kernel changes supporting this:
- Remove runRlnRefreshRetry from legacyLightpushPublish. The legacy
path now schedules the refresh and returns the error tagged with
RlnProofRefreshScheduledMsg, matching the non-legacy path; retrying
is the caller's decision. Drops the now-unused
RlnMerkleProofRefreshTimeout.
- generateRLNProofWithRootRefresh reuses the nonce drawn for the first
attempt when it regenerates after a stale root, rather than drawing a
second. Only the merkle path differs between the two attempts, so a
redraw would spend two message ids from the epoch budget on a single
message and drift the rate limit manager's accounting away from the
nonce manager's.
Adds Waku.isRlnRejection / Waku.onRlnProofRejected as the messaging
layer's handle on the kernel's RLN rejection detection and background
refresh. Updates the legacy lightpush tests to the schedule-refresh
contract.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* test: cover currentRlnEpochQuota (mounted and unmounted)
currentRlnEpochQuota ships in the enforcement PR, but its mounted-RLN
assertion needs the anvil-backed group-manager scaffolding that lives in
this file, so the coverage rides along here: none when RLN is unmounted,
and the epoch index + userMessageLimit when it is.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
* refactor: consolidate the RLN-rejection parking into parkForRlnProofRefresh
The relay and lightpush processors duplicated the RLN-rejection recovery
(schedule background refresh, clear the stale proof, reset admission,
park as NextRoundRetry); both now call parkForRlnProofRefresh in
send_processor, so the proof-clear the retry contract depends on cannot
drift between the two processors. Also resets firstAdmittedTime so the
regenerated proof's fresh nonce is re-admitted rather than sent uncharged.
No behavior change.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
* docs: clarify that lightpush reuses an already-attached RLN proof
The old wording ("attaches an RLN proof per attempt") reads as if every
retry redraws a nonce. The flow proves a message only when it carries no
proof, so a task admitted once reuses its proof and nonce across retries.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat: retry once on scheduled RLN proof refresh in the lightpush REST handlers
The kernel lightpush publish paths no longer retry an RLN-invalid publish
inline. On a stale merkle root they schedule a background cache refresh and
return early, tagging the error with RlnProofRefreshScheduledMsg — retrying
is the caller's decision, so the send service recovers through its own loop
and the kernel exposes mechanism only.
The synchronous REST endpoints have no such loop: they call publish once and
map the result to an HTTP status. Left unchanged, a transient stale-root
rejection that the kernel previously absorbed via runRlnRefreshRetry would
now surface to the HTTP client as a 503. Restore the transparent retry where
it belongs under this layering — at the caller — instead of back in the
kernel where it would re-nest inside the send service's retry.
Both the legacy and v3 handlers now retry the publish exactly once when the
first result carries RlnProofRefreshScheduledMsg, under the same
FutTimeoutForPushRequestProcessing bound. The handler's message carries no
proof, so the retry regenerates against the refreshed merkle path. Any other
error, and any error on the retry itself, maps to its response as before.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
* fix: retry RLN proof attach on a charged-but-unproven task
admitAndProve set firstAdmittedTime before attaching the proof, then
guarded its whole body on firstAdmittedTime.isSome(). A transient proof
attach failure (e.g. NonceLimitReached) left the task charged but with an
empty proof, and the next round's early-return skipped the attach entirely
and shipped the message bare. The docstring's own invariant — "once
admitted, a task keeps its slot and its proof" — was violated: it kept the
slot but not the proof.
Guard only the rate-limit charge on firstAdmittedTime, not the attach.
attachRlnProof is already idempotent (short-circuits when RLN is unmounted
or a proof is present), so it is safe to call every round: a charged-but-
unproven task retries the attach until it sticks, then short-circuits. The
ordering invariant holds (charge strictly before attach, so an over-budget
message never draws a nonce), the charge stays once-per-task, and NO_PEERS
retries remain free.
This also removes a latent relay double-charge: previously a bare message
reached the relay, was rejected as RLN-invalid, and parkForRlnProofRefresh
reset firstAdmittedTime — re-charging a slot on the next round. The message
now never leaves unproven.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Tanya S
GitHub
parent
4809b991df
commit
6a1af006b1
@ -33,6 +33,11 @@ method sendImpl*(
|
||||
await self.waku.lightpushPublishToAny(task.pubsubTopic, task.msg)
|
||||
).valueOr:
|
||||
error "LightpushSendProcessor.sendImpl failed", error = error.desc.get($error.code)
|
||||
|
||||
if error.isRlnRejection():
|
||||
task.parkForRlnProofRefresh(self.waku)
|
||||
return
|
||||
|
||||
case error.code
|
||||
of LightPushErrorCode.NO_PEERS_TO_RELAY, LightPushErrorCode.TOO_MANY_REQUESTS,
|
||||
LightPushErrorCode.OUT_OF_RLN_PROOF, LightPushErrorCode.SERVICE_NOT_AVAILABLE,
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import results, chronos, chronicles
|
||||
import brokers/broker_context
|
||||
import logos_delivery/waku/[waku_core], logos_delivery/waku/waku_lightpush/[common, rpc]
|
||||
import logos_delivery/waku/waku, logos_delivery/waku/api/publish
|
||||
import logos_delivery/waku/requests/health_requests
|
||||
import logos_delivery/api/types
|
||||
import ./[delivery_task, send_processor]
|
||||
@ -9,6 +10,7 @@ logScope:
|
||||
topics = "send service relay processor"
|
||||
|
||||
type RelaySendProcessor* = ref object of BaseSendProcessor
|
||||
waku: Waku
|
||||
publishProc: PushMessageHandler
|
||||
fallbackStateToSet: DeliveryState
|
||||
|
||||
@ -16,6 +18,7 @@ proc new*(
|
||||
T: typedesc[RelaySendProcessor],
|
||||
lightpushAvailable: bool,
|
||||
publishProc: PushMessageHandler,
|
||||
waku: Waku,
|
||||
brokerCtx: BrokerContext,
|
||||
): RelaySendProcessor =
|
||||
let fallbackStateToSet =
|
||||
@ -25,6 +28,7 @@ proc new*(
|
||||
DeliveryState.FailedToDeliver
|
||||
|
||||
return RelaySendProcessor(
|
||||
waku: waku,
|
||||
publishProc: publishProc,
|
||||
fallbackStateToSet: fallbackStateToSet,
|
||||
brokerCtx: brokerCtx,
|
||||
@ -60,6 +64,11 @@ method sendImpl*(self: RelaySendProcessor, task: DeliveryTask) {.async.} =
|
||||
let errorMessage = error.desc.get($error.code)
|
||||
error "Failed to publish message with relay",
|
||||
request = task.requestId, msgHash = task.msgHash.to0xHex(), error = errorMessage
|
||||
|
||||
if error.isRlnRejection():
|
||||
task.parkForRlnProofRefresh(self.waku)
|
||||
return
|
||||
|
||||
if error.code != LightPushErrorCode.NO_PEERS_TO_RELAY:
|
||||
task.state = DeliveryState.FailedToDeliver
|
||||
task.errorDesc = errorMessage
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
import chronos
|
||||
import results, chronos
|
||||
import brokers/broker_context
|
||||
import logos_delivery/waku/waku, logos_delivery/waku/api/publish
|
||||
import ./delivery_task
|
||||
|
||||
{.push raises: [].}
|
||||
@ -21,6 +22,19 @@ method sendImpl*(
|
||||
): Future[void] {.async, base.} =
|
||||
assert false, "Not implemented"
|
||||
|
||||
proc parkForRlnProofRefresh*(task: DeliveryTask, waku: Waku) =
|
||||
## The service refused the task's proof as RLN-invalid: the message itself is
|
||||
## fine, its proof went stale against a moved merkle root. Schedules a
|
||||
## background merkle-path refresh and clears the proof so the next round
|
||||
## regenerates one against the refreshed path — `attachRlnProof`
|
||||
## short-circuits on an existing proof, so without the clear the rejected
|
||||
## bytes would be resent until age-out. Resetting admission re-charges the
|
||||
## fresh nonce that regeneration draws.
|
||||
waku.onRlnProofRejected()
|
||||
task.msg.proof = @[]
|
||||
task.firstAdmittedTime = Opt.none(Moment)
|
||||
task.state = DeliveryState.NextRoundRetry
|
||||
|
||||
method process*(
|
||||
self: BaseSendProcessor, task: DeliveryTask
|
||||
): Future[void] {.async, base.} =
|
||||
|
||||
@ -68,7 +68,9 @@ proc setupSendProcessorChain(
|
||||
|
||||
if isRelayAvail:
|
||||
let publishProc = waku.relayPushHandler()
|
||||
processors.add(RelaySendProcessor.new(isLightPushAvail, publishProc, brokerCtx))
|
||||
processors.add(
|
||||
RelaySendProcessor.new(isLightPushAvail, publishProc, waku, brokerCtx)
|
||||
)
|
||||
if isLightPushAvail:
|
||||
processors.add(LightpushSendProcessor.new(waku, brokerCtx))
|
||||
|
||||
@ -254,18 +256,27 @@ proc evaluateAndCleanUp(self: SendService) =
|
||||
)
|
||||
)
|
||||
|
||||
proc admitOnce(self: SendService, task: DeliveryTask): Future[bool] {.async.} =
|
||||
## Charges the task's first transmission against the epoch budget, at most
|
||||
## once per task (`firstAdmittedTime`); retries then resend for free. Returns
|
||||
## false when the task must stay parked for a later epoch.
|
||||
if task.firstAdmittedTime.isSome():
|
||||
return true
|
||||
proc admitAndProve(self: SendService, task: DeliveryTask): Future[bool] {.async.} =
|
||||
## Gates a task's first transmission: charges one epoch slot, then attaches
|
||||
## an RLN proof — strictly in that order, so an over-budget message never
|
||||
## draws a nonce. The slot is charged at most once per task lifetime
|
||||
## (`firstAdmittedTime`); the proof attach is retried each round until it
|
||||
## sticks, then short-circuits, so a task charged but not yet proven never
|
||||
## ships bare. Returns false while the task must stay parked for a later round.
|
||||
if task.firstAdmittedTime.isNone():
|
||||
(await self.rateLimitManager.admit(task.msg.payload)).isOkOr:
|
||||
debug "over rate-limit budget, task waits for the epoch to roll",
|
||||
requestId = task.requestId, msgHash = task.msgHash.to0xHex()
|
||||
return false
|
||||
task.firstAdmittedTime = Opt.some(Moment.now())
|
||||
|
||||
(await self.rateLimitManager.admit(task.msg.payload)).isOkOr:
|
||||
debug "over rate-limit budget, task waits for the epoch to roll",
|
||||
requestId = task.requestId, msgHash = task.msgHash.to0xHex()
|
||||
## A no-op when RLN is not mounted, or when a prior round already attached a
|
||||
## proof; otherwise draws the nonce and attaches.
|
||||
task.msg = (await self.waku.attachRlnProof(task.msg)).valueOr:
|
||||
error "failed to attach RLN proof, retrying next round",
|
||||
requestId = task.requestId, error = error
|
||||
return false
|
||||
task.firstAdmittedTime = Opt.some(Moment.now())
|
||||
|
||||
return true
|
||||
|
||||
proc trySendMessages*(self: SendService) {.async.} =
|
||||
@ -273,7 +284,7 @@ proc trySendMessages*(self: SendService) {.async.} =
|
||||
|
||||
for task in tasksToSend:
|
||||
# Todo, check if it has any perf gain to run them concurrent...
|
||||
if not (await self.admitOnce(task)):
|
||||
if not (await self.admitAndProve(task)):
|
||||
continue
|
||||
await self.sendProcessor.process(task)
|
||||
|
||||
@ -304,7 +315,7 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} =
|
||||
error "SendService.send: failed to subscribe to content topic",
|
||||
contentTopic = task.msg.contentTopic, error = error
|
||||
|
||||
if not (await self.admitOnce(task)):
|
||||
if not (await self.admitAndProve(task)):
|
||||
info "SendService.send: parking task for a later round",
|
||||
requestId = task.requestId, msgHash = task.msgHash.to0xHex()
|
||||
task.state = DeliveryState.NextRoundRetry
|
||||
|
||||
@ -7,6 +7,7 @@
|
||||
## so the messaging layer never inspects `waku.node` directly.
|
||||
{.push raises: [].}
|
||||
|
||||
import std/[times, strutils]
|
||||
import results, chronos
|
||||
|
||||
import logos_delivery/waku/waku
|
||||
@ -38,8 +39,8 @@ proc hasLightpush*(self: Waku): bool =
|
||||
|
||||
proc relayPushHandler*(self: Waku): PushMessageHandler =
|
||||
## Builds the relay publish handler used by the send pipeline. Caller
|
||||
## ensures relay is mounted. RLN proof generation is handled client-side
|
||||
## in (legacy)lightpushPublish; this handler only validates and republishes.
|
||||
## ensures relay is mounted. The handler validates and republishes; the
|
||||
## proof is attached by the messaging layer via `attachRlnProof`.
|
||||
return getRelayPushHandler(self.node.wakuRelay)
|
||||
|
||||
proc currentRlnEpochQuota*(self: Waku): Opt[tuple[epochIndex, messageLimit: uint64]] =
|
||||
@ -53,6 +54,55 @@ proc currentRlnEpochQuota*(self: Waku): Opt[tuple[epochIndex, messageLimit: uint
|
||||
|
||||
return Opt.some((fromEpoch(self.node.rln.getCurrentEpoch()), uint64(limit)))
|
||||
|
||||
proc attachRlnProof*(
|
||||
self: Waku, message: WakuMessage
|
||||
): Future[Result[WakuMessage, string]] {.async.} =
|
||||
## Returns `message` carrying an RLN proof. A message that already has one is
|
||||
## returned untouched, so retrying a task neither redraws a nonce nor changes
|
||||
## the bytes. Without RLN mounted the message passes through unproven.
|
||||
##
|
||||
## Uses the root-refreshing generator: a message can wait in the send
|
||||
## service's task cache while the group root moves on chain, so the proof is
|
||||
## validated against the acceptable-root window and regenerated once against a
|
||||
## refetched merkle path if it went stale.
|
||||
if self.node.rln.isNil() or message.proof.len > 0:
|
||||
return ok(message)
|
||||
|
||||
var msgWithProof = message
|
||||
msgWithProof.proof = (
|
||||
await self.node.rln.generateRLNProofWithRootRefresh(
|
||||
message.toRLNSignal(), float64(getTime().toUnix())
|
||||
)
|
||||
).valueOr:
|
||||
return err("failed to attach RLN proof: " & error)
|
||||
|
||||
return ok(msgWithProof)
|
||||
|
||||
func isRlnRejection*(error: ErrorStatus): bool =
|
||||
## True when a publish failure means "the RLN proof was not accepted", so the
|
||||
## message is worth retrying with a freshly generated proof rather than being
|
||||
## failed outright.
|
||||
##
|
||||
## OUT_OF_RLN_PROOF is always RLN. INVALID_MESSAGE also covers non-RLN
|
||||
## rejections (an oversized message, say), so it additionally has to carry the
|
||||
## validator's error marker — this is the same gate the kernel lightpush path
|
||||
## applies before scheduling a refresh.
|
||||
return
|
||||
error.code == LightPushErrorCode.OUT_OF_RLN_PROOF or (
|
||||
error.code == LightPushErrorCode.INVALID_MESSAGE and
|
||||
error.desc.get("").contains(RlnValidatorErrorMsg)
|
||||
)
|
||||
|
||||
proc onRlnProofRejected*(self: Waku) =
|
||||
## Called when a publish was rejected as RLN-invalid. Starts refetching the
|
||||
## merkle path in the background, so the next proof generated for the message
|
||||
## is built against a fresh one. Non-blocking: the send service's own loop is
|
||||
## what retries, and it must not stall waiting on an RPC round trip.
|
||||
if self.node.rln.isNil():
|
||||
return
|
||||
|
||||
self.node.rln.groupManager.scheduleMerkleProofRefresh()
|
||||
|
||||
proc lightpushPeerAvailable*(self: Waku, shard: PubsubTopic): bool =
|
||||
## True if a lightpush service peer is available for `shard`.
|
||||
return self.node.peerManager.selectPeer(WakuLightPushCodec, Opt.some(shard)).isSome()
|
||||
@ -61,9 +111,9 @@ proc lightpushPublishToAny*(
|
||||
self: Waku, shard: PubsubTopic, message: WakuMessage
|
||||
): Future[WakuLightPushResult] {.async.} =
|
||||
## Selects a lightpush service peer for `shard` and publishes `message`
|
||||
## through the node's lightpush flow, which attaches an RLN proof per
|
||||
## attempt when RLN is mounted. Returns SERVICE_NOT_AVAILABLE when no peer
|
||||
## is available.
|
||||
## through the node's lightpush flow. With RLN mounted the flow proves
|
||||
## `message` only if it carries no proof, so an already-proven task reuses its
|
||||
## nonce. Returns SERVICE_NOT_AVAILABLE when no peer is available.
|
||||
let peer = self.node.peerManager.selectPeer(WakuLightPushCodec, Opt.some(shard)).valueOr:
|
||||
return lightpushResultServiceUnavailable("no lightpush peer available for shard")
|
||||
try:
|
||||
|
||||
@ -105,30 +105,6 @@ proc resolveLegacyPubsubTopic(
|
||||
return err("Autosharding error: " & error)
|
||||
return ok($shard)
|
||||
|
||||
proc runRlnRefreshRetry(
|
||||
node: WakuNode,
|
||||
rln: Opt[Rln],
|
||||
msgWithProof: WakuMessage,
|
||||
pubsubForPublish: PubsubTopic,
|
||||
peer: RemotePeerInfo,
|
||||
fallback: legacy_lightpush_protocol.WakuLightPushResult[string],
|
||||
): Future[legacy_lightpush_protocol.WakuLightPushResult[string]] {.async, gcsafe.} =
|
||||
## Refreshes the RLN merkle proof path and retries the publish once. Only the
|
||||
## refresh is bounded by RlnMerkleProofRefreshTimeout (returning `fallback` on
|
||||
## timeout); the retried publish runs unbounded, matching the first attempt.
|
||||
info "legacy lightpush send rejected as RLN-invalid; " &
|
||||
"refreshing merkle proof and retrying once"
|
||||
rln.get().groupManager.invalidateMerkleProofCache()
|
||||
|
||||
let refreshFut = attachRLNProof(rln.get(), msgWithProof)
|
||||
if not (await refreshFut.withTimeout(RlnMerkleProofRefreshTimeout)):
|
||||
warn "legacy lightpush RLN proof refresh timed out; returning original error"
|
||||
return fallback
|
||||
let retryMsg = refreshFut.read().valueOr:
|
||||
return err("failed call attachRLNProof from lightpush retry: " & error)
|
||||
|
||||
return await internalLegacyLightpushPublish(node, pubsubForPublish, retryMsg, peer)
|
||||
|
||||
proc legacyLightpushPublish*(
|
||||
node: WakuNode,
|
||||
pubsubTopic: Opt[PubsubTopic],
|
||||
@ -161,18 +137,20 @@ proc legacyLightpushPublish*(
|
||||
).valueOr:
|
||||
return err(error)
|
||||
|
||||
let firstResult =
|
||||
let publishResult =
|
||||
await internalLegacyLightpushPublish(node, pubsubForPublish, msgWithProof, peer)
|
||||
|
||||
# Legacy has no status codes, so string-match the RLN error to detect a
|
||||
# stale merkle proof path, then refresh and retry once.
|
||||
if firstResult.isOk() or rln.isNone() or
|
||||
not firstResult.error.contains(RlnValidatorErrorMsg):
|
||||
return firstResult
|
||||
# stale merkle proof path. Schedule the refresh and hand the error back:
|
||||
# retrying is the caller's decision, the same way the non-legacy path
|
||||
# behaves. A retry regenerates the proof against the refreshed cache.
|
||||
if publishResult.isOk() or rln.isNone() or
|
||||
not publishResult.error.contains(RlnValidatorErrorMsg):
|
||||
return publishResult
|
||||
|
||||
return await runRlnRefreshRetry(
|
||||
node, rln, msgWithProof, pubsubForPublish, peer, firstResult
|
||||
)
|
||||
info "legacy lightpush send rejected as RLN-invalid; scheduling merkle proof refresh"
|
||||
rln.get().groupManager.scheduleMerkleProofRefresh()
|
||||
return err(RlnProofRefreshScheduledMsg & ": " & publishResult.error)
|
||||
except CatchableError:
|
||||
return err(getCurrentExceptionMsg())
|
||||
|
||||
|
||||
@ -3,6 +3,7 @@
|
||||
import
|
||||
std/strformat,
|
||||
std/options,
|
||||
std/strutils,
|
||||
results,
|
||||
stew/byteutils,
|
||||
chronicles,
|
||||
@ -15,6 +16,7 @@ import
|
||||
logos_delivery/waku/node/peer_manager,
|
||||
logos_delivery/waku/waku_lightpush_legacy/common,
|
||||
../../../waku_node,
|
||||
../../../rln,
|
||||
../../handlers,
|
||||
../serdes,
|
||||
../responses,
|
||||
@ -72,13 +74,27 @@ proc installLightPushRequestHandler*(
|
||||
peerOp.valueOr:
|
||||
return NoPeerNoneFoundError
|
||||
|
||||
let subFut = node.legacyLightpushPublish(req.pubsubTopic, msg, peer)
|
||||
var pushFut = node.legacyLightpushPublish(req.pubsubTopic, msg, peer)
|
||||
|
||||
if not await subFut.withTimeout(FutTimeoutForPushRequestProcessing):
|
||||
if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing):
|
||||
error "Failed to request a message push due to timeout!"
|
||||
return RestApiResponse.serviceUnavailable("Push request timed out")
|
||||
|
||||
subFut.value().isOkOr:
|
||||
var pushResult = pushFut.value()
|
||||
|
||||
# An error tagged RlnProofRefreshScheduledMsg is a publish rejected on a
|
||||
# stale merkle root. On this error the kernel scheduled a cache refresh
|
||||
# before failing early. This synchronous endpoint has no retry loop of its
|
||||
# own, so retry once — the second attempt generates its proof against the
|
||||
# refreshed merkle path.
|
||||
if pushResult.isErr() and pushResult.error.contains(RlnProofRefreshScheduledMsg):
|
||||
pushFut = node.legacyLightpushPublish(req.pubsubTopic, msg, peer)
|
||||
if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing):
|
||||
error "Failed to request a message push due to timeout!"
|
||||
return RestApiResponse.serviceUnavailable("Push request timed out")
|
||||
pushResult = pushFut.value()
|
||||
|
||||
pushResult.isOkOr:
|
||||
if error == TooManyRequestsMessage:
|
||||
return RestApiResponse.tooManyRequests("Request rate limmit reached")
|
||||
|
||||
|
||||
@ -3,6 +3,7 @@
|
||||
import
|
||||
std/strformat,
|
||||
std/options,
|
||||
std/strutils,
|
||||
results,
|
||||
stew/byteutils,
|
||||
chronicles,
|
||||
@ -15,6 +16,7 @@ import
|
||||
logos_delivery/waku/node/peer_manager,
|
||||
logos_delivery/waku/waku_lightpush/common,
|
||||
../../../waku_node,
|
||||
../../../rln,
|
||||
../../handlers,
|
||||
../serdes,
|
||||
../responses,
|
||||
@ -96,11 +98,27 @@ proc installLightPushRequestHandler*(
|
||||
makeRestResponse(lightpushResultServiceUnavailable(NoPeerNoneFoundError))
|
||||
toPeer = Opt.some(aPeer)
|
||||
|
||||
let subFut = node.lightpushPublish(req.pubsubTopic, msg, toPeer)
|
||||
var pushFut = node.lightpushPublish(req.pubsubTopic, msg, toPeer)
|
||||
|
||||
if not await subFut.withTimeout(FutTimeoutForPushRequestProcessing):
|
||||
if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing):
|
||||
error "Failed to request a message push due to timeout!"
|
||||
return
|
||||
makeRestResponse(lightpushResultServiceUnavailable("Push request timed out"))
|
||||
|
||||
return makeRestResponse(subFut.value())
|
||||
var pushResult = pushFut.value()
|
||||
|
||||
# An error tagged RlnProofRefreshScheduledMsg is a publish rejected on a
|
||||
# stale merkle root. On this error the kernel scheduled a cache refresh
|
||||
# before failing early. This synchronous endpoint has no retry loop of its
|
||||
# own, so retry once — the second attempt generates its proof against the
|
||||
# refreshed merkle path.
|
||||
if pushResult.isErr() and
|
||||
pushResult.error.desc.get("").contains(RlnProofRefreshScheduledMsg):
|
||||
pushFut = node.lightpushPublish(req.pubsubTopic, msg, toPeer)
|
||||
if not await pushFut.withTimeout(FutTimeoutForPushRequestProcessing):
|
||||
error "Failed to request a message push due to timeout!"
|
||||
return
|
||||
makeRestResponse(lightpushResultServiceUnavailable("Push request timed out"))
|
||||
pushResult = pushFut.value()
|
||||
|
||||
return makeRestResponse(pushResult)
|
||||
|
||||
@ -25,11 +25,6 @@ const RlnValidatorErrorMsg* = "RLN validation failed"
|
||||
const RlnProofRefreshScheduledMsg* =
|
||||
"stale RLN proof suspected; refresh scheduled, retry the publish"
|
||||
|
||||
# Bounds the legacy lightpush merkle proof refresh (eth_call refetch + proof
|
||||
# regen) so a hanging RPC cannot stall the caller. The retried publish is not
|
||||
# bounded.
|
||||
const RlnMerkleProofRefreshTimeout* = 5.seconds
|
||||
|
||||
# inputs of the membership contract constructor
|
||||
# TODO may be able to make these constants private and put them inside the waku_rln_utils
|
||||
const
|
||||
|
||||
@ -60,12 +60,14 @@ proc toRLNSignal*(wakumessage: WakuMessage): seq[byte] =
|
||||
output = concat(wakumessage.payload, contentTopicBytes, @(timestampBytes))
|
||||
return output
|
||||
|
||||
proc generateRLNProof*(
|
||||
rln: Rln, input: seq[byte], senderEpochTime: float64
|
||||
proc generateRLNProofWithNonce(
|
||||
rln: Rln, input: seq[byte], senderEpochTime: float64, nonce: Nonce
|
||||
): Future[Result[seq[byte], string]] {.async: (raises: []).} =
|
||||
## Generates a proof against an already drawn `nonce`. Regenerating for an
|
||||
## unchanged (input, epoch, nonce) is safe: the revealed share is a function
|
||||
## of those three, so a regenerated proof reveals the same share and cannot
|
||||
## read as double-signalling.
|
||||
let epoch = rln.calcEpoch(senderEpochTime)
|
||||
let nonce = rln.nonceManager.getNonce().valueOr:
|
||||
return err("could not get new message id to generate an rln proof: " & $error)
|
||||
try:
|
||||
let proof = (await rln.groupManager.generateProof(input, epoch, nonce)).valueOr:
|
||||
return err("could not generate rln-v2 proof: " & $error)
|
||||
@ -73,13 +75,29 @@ proc generateRLNProof*(
|
||||
except CatchableError as e:
|
||||
return err("exception generating rln proof: " & e.msg)
|
||||
|
||||
proc generateRLNProof*(
|
||||
rln: Rln, input: seq[byte], senderEpochTime: float64
|
||||
): Future[Result[seq[byte], string]] {.async: (raises: []).} =
|
||||
let nonce = rln.nonceManager.getNonce().valueOr:
|
||||
return err("could not get new message id to generate an rln proof: " & $error)
|
||||
return await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce)
|
||||
|
||||
proc generateRLNProofWithRootRefresh*(
|
||||
rln: Rln, input: seq[byte], senderEpochTime: float64
|
||||
): Future[Result[seq[byte], string]] {.async.} =
|
||||
## Generates an RLN proof and checks its merkle root against the
|
||||
## acceptable-root window. If the root is stale, invalidates the cache and
|
||||
## regenerates once against a refetched path. Returns the proof bytes.
|
||||
let proofBytes = (await rln.generateRLNProof(input, senderEpochTime)).valueOr:
|
||||
##
|
||||
## The regeneration reuses the nonce drawn for the first attempt: only the
|
||||
## merkle path differs between the two, so drawing again would spend two
|
||||
## message ids from the epoch budget on a message that is sent once. That
|
||||
## would drift the budget the rate limit manager accounts for away from the
|
||||
## one the nonce manager enforces.
|
||||
let nonce = rln.nonceManager.getNonce().valueOr:
|
||||
return err("could not get new message id to generate an rln proof: " & $error)
|
||||
|
||||
let proofBytes = (await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce)).valueOr:
|
||||
return err("failed to generate RLN proof: " & $error)
|
||||
|
||||
let rlnProof = RateLimitProof.init(proofBytes).valueOr:
|
||||
@ -90,7 +108,7 @@ proc generateRLNProofWithRootRefresh*(
|
||||
|
||||
info "RLN: stale merkle root detected; refreshing merkle path and regenerating proof"
|
||||
rln.groupManager.invalidateMerkleProofCache()
|
||||
return await rln.generateRLNProof(input, senderEpochTime)
|
||||
return await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce)
|
||||
|
||||
proc attachRLNProof*(
|
||||
r: Rln, message: WakuMessage
|
||||
|
||||
@ -1,4 +1,7 @@
|
||||
{.used.}
|
||||
|
||||
import
|
||||
./test_rate_limit_manager, ./test_delivery_task_reaping, ./test_send_service_scheduler
|
||||
./test_rate_limit_manager,
|
||||
./test_rln_proof_attach,
|
||||
./test_delivery_task_reaping,
|
||||
./test_send_service_scheduler
|
||||
|
||||
116
tests/messaging/test_rln_proof_attach.nim
Normal file
116
tests/messaging/test_rln_proof_attach.nim
Normal file
@ -0,0 +1,116 @@
|
||||
{.used.}
|
||||
|
||||
import std/[options, net, osproc]
|
||||
import chronos, testutils/unittests, results, stew/byteutils
|
||||
import
|
||||
logos_delivery/waku/[waku, waku_core, rln],
|
||||
logos_delivery/waku/node/waku_node,
|
||||
logos_delivery/waku/node/waku_node/relay,
|
||||
logos_delivery/waku/api/publish,
|
||||
logos_delivery/api/conf/messaging_conf,
|
||||
logos_delivery/waku/factory/waku_conf
|
||||
import
|
||||
../testlib/testasync,
|
||||
../waku_rln_relay/utils_onchain,
|
||||
../waku_rln_relay/rln/waku_rln_relay_utils
|
||||
|
||||
proc testConf(): WakuConf =
|
||||
var conf = MessagingClientConf()
|
||||
.toWakuNodeConf(messaging_conf.LogosDeliveryMode.Core).valueOr:
|
||||
raiseAssert error
|
||||
conf.listenAddress = parseIpAddress("0.0.0.0")
|
||||
conf.tcpPort = Port(0)
|
||||
conf.discv5UdpPort = Port(0)
|
||||
conf.clusterId = Opt.some(3'u16)
|
||||
conf.numShardsInNetwork = 1
|
||||
conf.rest = false
|
||||
return conf.toWakuConf().valueOr:
|
||||
raiseAssert error
|
||||
|
||||
proc testMessage(): WakuMessage =
|
||||
WakuMessage(
|
||||
payload: "hello".toBytes(),
|
||||
contentTopic: "/test/1/attach/proto",
|
||||
timestamp: 1_700_000_000_000_000_000,
|
||||
)
|
||||
|
||||
suite "SendService RLN proof attach":
|
||||
asyncTest "passes the message through unproven when RLN is not mounted":
|
||||
## The default (no-RLN) configuration must be unaffected: no proof is
|
||||
## attached and the message reaches the send processors unchanged.
|
||||
let waku = (await Waku.new(testConf())).expect("Waku.new")
|
||||
let msg = testMessage()
|
||||
|
||||
let attached = (await waku.attachRlnProof(msg)).expect("attachRlnProof")
|
||||
|
||||
check:
|
||||
attached.proof.len == 0
|
||||
attached.payload == msg.payload
|
||||
attached.contentTopic == msg.contentTopic
|
||||
|
||||
asyncTest "currentRlnEpochQuota is none when RLN is not mounted":
|
||||
## The rate limit manager reads `none` as "use the wall-clock fallback".
|
||||
let waku = (await Waku.new(testConf())).expect("Waku.new")
|
||||
check waku.currentRlnEpochQuota().isNone()
|
||||
|
||||
suite "SendService RLN proof attach - RLN mounted":
|
||||
var
|
||||
waku {.threadvar.}: Waku
|
||||
anvilProc {.threadvar.}: Process
|
||||
manager {.threadvar.}: OnchainGroupManager
|
||||
|
||||
asyncSetup:
|
||||
anvilProc = runAnvil(stateFile = Opt.some(DEFAULT_ANVIL_STATE_PATH))
|
||||
manager = waitFor setupOnchainGroupManager(deployContracts = false)
|
||||
|
||||
waku = (await Waku.new(testConf())).expect("Waku.new")
|
||||
await waku.node.setRlnValidator(
|
||||
getWakuRlnConfig(
|
||||
manager = manager,
|
||||
userMessageLimit = 20,
|
||||
index = MembershipIndex(1),
|
||||
epochSizeSec = 600,
|
||||
)
|
||||
)
|
||||
|
||||
let credentials = generateCredentials()
|
||||
(
|
||||
waitFor cast[OnchainGroupManager](waku.node.rln.groupManager).register(
|
||||
credentials, UserMessageLimit(20)
|
||||
)
|
||||
).isOkOr:
|
||||
assert false, "failed to register RLN credentials: " & error
|
||||
|
||||
asyncTeardown:
|
||||
## The RLN proof-generator provider is registered on the global broker
|
||||
## context; without stopping RLN it leaks into the next test's setup.
|
||||
try:
|
||||
await waku.node.rln.stop()
|
||||
except Exception:
|
||||
assert false, "failed to stop RLN: " & getCurrentExceptionMsg()
|
||||
stopAnvil(anvilProc)
|
||||
|
||||
asyncTest "attaches a proof":
|
||||
let attached = (await waku.attachRlnProof(testMessage())).expect("attachRlnProof")
|
||||
|
||||
check attached.proof.len > 0
|
||||
|
||||
asyncTest "currentRlnEpochQuota reports RLN's epoch and user message limit":
|
||||
## Wires the rate limit manager to RLN: the manager clamps its configured
|
||||
## cap to `messageLimit` and rolls on `epochIndex`.
|
||||
let quota = waku.currentRlnEpochQuota()
|
||||
check:
|
||||
quota.isSome()
|
||||
quota.get().messageLimit == 20'u64 # the mounted userMessageLimit
|
||||
quota.get().epochIndex > 0'u64 # unixTime div epochSize, far from zero
|
||||
|
||||
asyncTest "is idempotent: a message that already carries a proof is untouched":
|
||||
## Pins the retry contract: the send service re-attaches on every round, so
|
||||
## re-attaching must neither draw a fresh nonce nor change the bytes —
|
||||
## otherwise a retried task would resend under a new nullifier.
|
||||
let first = (await waku.attachRlnProof(testMessage())).expect("first attach")
|
||||
let second = (await waku.attachRlnProof(first)).expect("second attach")
|
||||
|
||||
check:
|
||||
first.proof.len > 0
|
||||
second.proof == first.proof
|
||||
@ -187,21 +187,24 @@ suite "RLN Proofs as a Lightpush Service":
|
||||
|
||||
# The tests below drive `server.legacyLightpushPublish(...)` against the
|
||||
# server node. Because `server.wakuLegacyLightPush` is mounted (and no
|
||||
# legacy client is), the call takes the self-request path — it still runs
|
||||
# the full client-side flow (proof gen, retry on RlnValidatorErrorMsg
|
||||
# substring, one-retry cap), but the request lands in the local
|
||||
# legacy client is), the call takes the self-request path — it runs the
|
||||
# full client-side flow (proof gen, RLN-rejection detection via the
|
||||
# RlnValidatorErrorMsg substring), but the request lands in the local
|
||||
# pushHandler. Swapping in a stub pushHandler lets each test control what
|
||||
# attempt N sees.
|
||||
# the publish attempt sees.
|
||||
#
|
||||
# On an RLN rejection the publish schedules a background merkle-proof
|
||||
# refresh and returns the error tagged with RlnProofRefreshScheduledMsg;
|
||||
# the caller (the send service loop) regenerates the proof and republishes
|
||||
# on its next round.
|
||||
|
||||
asyncTest "retry fires on RlnValidatorErrorMsg substring and second attempt succeeds":
|
||||
asyncTest "RLN rejection schedules a refresh and surfaces the tagged error":
|
||||
var callCount = 0
|
||||
let stub: PushMessageHandler = proc(
|
||||
pubsubTopic: PubsubTopic, message: WakuMessage
|
||||
): Future[WakuLightPushResult[void]] {.async.} =
|
||||
inc callCount
|
||||
if callCount == 1:
|
||||
return err(RlnValidatorErrorMsg & ": simulated stale merkle path")
|
||||
return ok()
|
||||
return err(RlnValidatorErrorMsg & ": simulated stale merkle path")
|
||||
server.wakuLegacyLightPush.pushHandler = stub
|
||||
|
||||
let response = await server.legacyLightpushPublish(
|
||||
@ -209,8 +212,10 @@ suite "RLN Proofs as a Lightpush Service":
|
||||
)
|
||||
|
||||
check:
|
||||
callCount == 2
|
||||
response.isOk()
|
||||
callCount == 1
|
||||
response.isErr()
|
||||
response.error.contains(RlnProofRefreshScheduledMsg)
|
||||
response.error.contains(RlnValidatorErrorMsg)
|
||||
|
||||
asyncTest "no retry when error does not contain RlnValidatorErrorMsg":
|
||||
var callCount = 0
|
||||
@ -230,27 +235,9 @@ suite "RLN Proofs as a Lightpush Service":
|
||||
response.isErr()
|
||||
response.error == "unrelated failure"
|
||||
|
||||
asyncTest "retry cap: two consecutive RLN errors surface the second":
|
||||
var callCount = 0
|
||||
let stub: PushMessageHandler = proc(
|
||||
pubsubTopic: PubsubTopic, message: WakuMessage
|
||||
): Future[WakuLightPushResult[void]] {.async.} =
|
||||
inc callCount
|
||||
return err(RlnValidatorErrorMsg & ": still stale")
|
||||
server.wakuLegacyLightPush.pushHandler = stub
|
||||
|
||||
let response = await server.legacyLightpushPublish(
|
||||
Opt.some(pubsubTopic), message, server.peerInfo.toRemotePeerInfo()
|
||||
)
|
||||
|
||||
check:
|
||||
callCount == 2
|
||||
response.isErr()
|
||||
response.error.contains(RlnValidatorErrorMsg)
|
||||
|
||||
asyncTest "no retry when node.rln is nil":
|
||||
# Detach RLN so the retry branch short-circuits on rln.isNone() even
|
||||
# when the error string carries RlnValidatorErrorMsg. Restore before
|
||||
asyncTest "no refresh scheduled when node.rln is nil":
|
||||
# Detach RLN so the RLN-rejection branch short-circuits on rln.isNone()
|
||||
# even when the error string carries RlnValidatorErrorMsg. Restore before
|
||||
# teardown so server.stop() sees the same object graph it was
|
||||
# constructed with.
|
||||
let savedRln = server.rln
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user