diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index 32ab9b235..844186057 100644 --- a/logos_delivery/api/conf/messaging_conf.nim +++ b/logos_delivery/api/conf/messaging_conf.nim @@ -6,9 +6,9 @@ import libp2p/crypto/crypto import logos_delivery/api/conf/kernel_conf import logos_delivery/waku/common/logging import logos_delivery/waku/factory/networks_config -import logos_delivery/messaging/rate_limit_manager/rate_limit_manager +import logos_delivery/messaging/rate_limit_manager/rate_limit_config -export kernel_conf, rate_limit_manager +export kernel_conf, rate_limit_config type LogosDeliveryMode* {.pure.} = enum Edge # client-only node @@ -56,8 +56,8 @@ type MessagingClientConf* = object nodeKey* {.name: "nodekey".}: Option[crypto.PrivateKey] ## P2P node private key (64-char hex): stable identity / peerId across restarts. rateLimit*: RateLimitConfig = RateLimitConfig( - epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch - ) ## RLN-epoch transmission budget enforced by the send service. + epochSizeSec: DefaultEpochSizeSec, userMessageLimit: DefaultUserMessageLimit + ) ## Per-epoch user message limit enforced by the send service. proc applyMode*(conf: var WakuNodeConf, mode: LogosDeliveryMode): ConfResult[void] = ## Sets the protocol flags implied by the mode. diff --git a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim index fa199774b..554b5e166 100644 --- a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim +++ b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim @@ -36,6 +36,17 @@ method sendImpl*( await self.waku.lightpushPublishToAny(task.pubsubTopic, task.msg) ).valueOr: error "LightpushSendProcessor.sendImpl failed", error = error.desc.get($error.code) + + if error.isRlnRejection(): + ## The proof was refused, so it must not be sent again: drop it and let + ## the refreshed merkle path produce a new one on the next round. + ## Re-admission gates the regeneration, so a task cannot spin through the + ## epoch budget by retrying. + self.waku.onRlnProofRejected() + task.msg.proof = @[] + task.state = DeliveryState.NextRoundRetry + return + case error.code of LightPushErrorCode.NO_PEERS_TO_RELAY, LightPushErrorCode.TOO_MANY_REQUESTS, LightPushErrorCode.OUT_OF_RLN_PROOF, LightPushErrorCode.SERVICE_NOT_AVAILABLE, diff --git a/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim b/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim index dc40b797d..3dd2a3aaa 100644 --- a/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim +++ b/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim @@ -3,6 +3,7 @@ import std/options import chronos, chronicles import brokers/broker_context import logos_delivery/waku/[waku_core], logos_delivery/waku/waku_lightpush/[common, rpc] +import logos_delivery/waku/waku, logos_delivery/waku/api/publish import logos_delivery/waku/requests/health_requests import logos_delivery/api/types import ./[delivery_task, send_processor] @@ -11,6 +12,7 @@ logScope: topics = "send service relay processor" type RelaySendProcessor* = ref object of BaseSendProcessor + waku: Waku publishProc: PushMessageHandler fallbackStateToSet: DeliveryState @@ -18,6 +20,7 @@ proc new*( T: typedesc[RelaySendProcessor], lightpushAvailable: bool, publishProc: PushMessageHandler, + waku: Waku, brokerCtx: BrokerContext, ): RelaySendProcessor = let fallbackStateToSet = @@ -27,6 +30,7 @@ proc new*( DeliveryState.FailedToDeliver return RelaySendProcessor( + waku: waku, publishProc: publishProc, fallbackStateToSet: fallbackStateToSet, brokerCtx: brokerCtx, @@ -62,6 +66,17 @@ method sendImpl*(self: RelaySendProcessor, task: DeliveryTask) {.async.} = let errorMessage = error.desc.get($error.code) error "Failed to publish message with relay", request = task.requestId, msgHash = task.msgHash.to0xHex(), error = errorMessage + + if error.isRlnRejection(): + ## The relay validator refused the proof. Dropping it and retrying is not + ## the same as failing: the message is valid, its proof went stale against + ## a moved merkle root. Clearing it makes the next round regenerate one + ## against the refreshed path. + self.waku.onRlnProofRejected() + task.msg.proof = @[] + task.state = DeliveryState.NextRoundRetry + return + if error.code != LightPushErrorCode.NO_PEERS_TO_RELAY: task.state = DeliveryState.FailedToDeliver task.errorDesc = errorMessage diff --git a/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim index ce2d9ead5..c6c774ffd 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim @@ -69,7 +69,9 @@ proc setupSendProcessorChain( if isRelayAvail: let publishProc = waku.relayPushHandler() - processors.add(RelaySendProcessor.new(isLightPushAvail, publishProc, brokerCtx)) + processors.add( + RelaySendProcessor.new(isLightPushAvail, publishProc, waku, brokerCtx) + ) if isLightPushAvail: processors.add(LightpushSendProcessor.new(waku, brokerCtx)) @@ -260,6 +262,15 @@ proc trySendMessages(self: SendService) {.async.} = ## `NextRoundRetry` and are retried as the epoch rolls over. (await self.rateLimitManager.admit(task.msg.payload)).isOkOr: continue + + ## Strictly after admission, so a rejected message never draws a nonce. + ## A no-op when RLN is not mounted, or when a prior round already + ## attached a proof. + task.msg = (await self.waku.attachRlnProof(task.msg)).valueOr: + error "SendService: failed to attach RLN proof, retrying next round", + requestId = task.requestId, error = error + continue + await self.sendProcessor.process(task) proc serviceLoop(self: SendService) {.async.} = @@ -296,6 +307,15 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} = self.addTask(task) return + ## Strictly after admission, so a rejected message never draws a nonce. + ## A no-op when RLN is not mounted. + task.msg = (await self.waku.attachRlnProof(task.msg)).valueOr: + error "SendService.send: failed to attach RLN proof, parking task", + requestId = task.requestId, error = error + task.state = DeliveryState.NextRoundRetry + self.addTask(task) + return + await self.sendProcessor.process(task) reportTaskResult(self, task) if task.state != DeliveryState.FailedToDeliver: diff --git a/logos_delivery/messaging/rate_limit_manager/quota_source.nim b/logos_delivery/messaging/rate_limit_manager/quota_source.nim new file mode 100644 index 000000000..763e7bc16 --- /dev/null +++ b/logos_delivery/messaging/rate_limit_manager/quota_source.nim @@ -0,0 +1,38 @@ +## Epoch + user-message-limit source for the rate limit manager. +## +## "Quota" is the tracking issues' word (#3838: "we should know our remaining +## quota") for what RLN expresses as an epoch's user message limit: each epoch +## grants `userMessageLimit` message ids (nonces), and the epoch rolling over +## refills the allowance. +## +## The manager rate-limits per epoch, but must not know *where* the epoch and +## the limit come from. With RLN mounted both are RLN's; without it, a +## wall-clock window and the configured limit stand in. Epoch and limit are +## read together, through one provider, so a read cannot straddle an epoch +## boundary and pair a fresh epoch with a stale limit. The provider is a +## callback so the manager stays free of any dependency on the `Waku` kernel — +## the RLN-backed provider is built one layer up, where the kernel handle is in +## scope. It returns `none` when RLN is not mounted, which is the signal to +## fall back to the wall clock. + +import std/[options, times] + +type + EpochQuota* = object + epochIndex*: uint64 + ## Current epoch as its numeric value (`timestamp div epochSizeSec`); + ## the kernel's `Epoch` type is the same value serialized to 32 bytes. + userMessageLimit*: uint64 + ## Message ids the epoch grants — the hard ceiling on admissions before + ## the epoch rolls over. + + QuotaProvider* = proc(): Option[EpochQuota] {.gcsafe, raises: [].} + ## The epoch and its user message limit, or `none` when RLN is not mounted. + +proc wallClockEpochIndex*(epochSizeSec: uint64): uint64 = + ## Absolute wall-clock epoch: `unixTime div epochSizeSec`. Absolute (not a + ## sliding window anchored at first use) so it matches RLN's `calcEpoch`, and + ## so two managers started at different moments agree on the boundary. + ## `epochSizeSec` is assumed positive; the manager only calls this when + ## enforcing. + uint64(getTime().toUnix()) div epochSizeSec diff --git a/logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim b/logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim new file mode 100644 index 000000000..f225ade4e --- /dev/null +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim @@ -0,0 +1,31 @@ +## Configuration for the messaging rate limit manager. +## +## Field names follow RLN terminology (`epochSizeSec`, `userMessageLimit` — see +## `RlnConf` and the RLN v2 spec) since they configure the local mirror of the +## same per-epoch limit RLN enforces on the network. +## +## Kept separate from the enforcement engine so the API config layer can depend +## on the vocabulary (`RateLimitConfig`) without pulling in the manager and its +## RLN seam. + +type + RateLimitError* {.pure.} = enum + OverBudget + + RateLimitConfig* = object + enabled*: bool ## spec: rate limiting opt-in; SHOULD be true when RLN active + epochSizeSec*: uint64 + ## Epoch length, in seconds. Only shapes the wall-clock fallback window; + ## when the RLN quota source is wired in it is ignored (the size is RLN's). + userMessageLimit*: uint64 + ## Messages allowed per epoch. When RLN is mounted the effective limit is + ## clamped to RLN's own user message limit; config can only tighten it. + +const + DefaultEpochSizeSec* = 600'u64 + DefaultUserMessageLimit* = 1'u64 + +func isEnforcing*(config: RateLimitConfig): bool = + ## Whether the config asks for actual rate limiting. A disabled or zeroed + ## configuration admits everything, so the manager can short-circuit. + config.enabled and config.epochSizeSec > 0 and config.userMessageLimit > 0 diff --git a/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim index 7ee197d8a..1ca57ad1a 100644 --- a/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim @@ -1,54 +1,81 @@ ## Rate Limit Manager for the Messaging API. ## -## Tracks messages sent per RLN epoch and rejects admission when the -## limit is approached, ensuring RLN compliance on enforcing relays. +## Rate-limits message transmissions against the per-epoch user message limit +## and rejects admission once it is spent, keeping the node within the quota +## that RLN-enforcing relays check. Each admission mirrors one RLN message id +## (nonce) draw; the epoch rolling over refills the allowance. ## -## For the skeleton this is a pass-through: every call is admitted. -## Real per-epoch budgeting will use `queue`, `currentEpochStart`, -## `sentInCurrentEpoch`, and `resetEpoch` to park messages and admit -## them as the epoch rolls over. +## The epoch and the limit come from a `QuotaProvider` when RLN is mounted +## (see `quota_source`); otherwise a wall-clock window and the configured limit +## stand in. Parking and retrying over-budget messages is the send service +## scheduler's job — this module only answers whether one more transmission +## fits the current epoch. ## ## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html -import std/times +import std/options import results, chronos -type - RateLimitError* {.pure.} = enum - OverBudget +import ./rate_limit_config, ./quota_source - RateLimitConfig* = object - enabled*: bool ## spec: rate limiting opt-in; SHOULD be true when RLN active - epochPeriodSec*: int - messagesPerEpoch*: int +export rate_limit_config, quota_source - RateLimitManager* = ref object - config*: RateLimitConfig - queue*: seq[seq[byte]] - currentEpochStart*: Time - sentInCurrentEpoch*: int +type RateLimitManager* = ref object + config*: RateLimitConfig + quotaProvider: QuotaProvider + ## Supplies the RLN epoch + user message limit. Nil, or a call returning + ## `none`, selects the wall-clock fallback. Queried per admission so a node + ## whose RLN mounts after construction upgrades automatically. + currentEpochIndex*: uint64 + sentInCurrentEpoch*: uint64 -const - DefaultEpochPeriodSec* = 600 - DefaultMessagesPerEpoch* = 1 +proc new*( + T: type RateLimitManager, + config: RateLimitConfig, + quotaProvider: QuotaProvider = nil, +): T = + return T( + config: config, + quotaProvider: quotaProvider, + currentEpochIndex: 0, + sentInCurrentEpoch: 0, + ) -proc new*(T: type RateLimitManager, config: RateLimitConfig): T = - return - T(config: config, queue: @[], currentEpochStart: getTime(), sentInCurrentEpoch: 0) +proc currentQuota(self: RateLimitManager): Option[EpochQuota] = + if self.quotaProvider.isNil(): + return none(EpochQuota) + return self.quotaProvider() proc admit*( self: RateLimitManager, msg: seq[byte] ): Future[Result[void, RateLimitError]] {.async: (raises: []).} = - ## Skeleton behaviour: admits immediately. Real per-epoch budgeting - ## will consult `config`, `sentInCurrentEpoch`, and the elapsed - ## `epochPeriodSec` window before admitting or parking `msg`. + ## Charges one message against the current epoch's user message limit, + ## rolling the window first when the epoch has advanced. A disabled or zeroed + ## configuration admits everything. + if not self.config.isEnforcing(): + return ok() + + let quota = self.currentQuota() + + let epochIndex = + if quota.isSome(): + quota.get().epochIndex + else: + wallClockEpochIndex(self.config.epochSizeSec) + + # RLN can only tighten the configured limit, never widen it: exceeding RLN's + # limit would fail later at proof generation, once the epoch's message ids + # are exhausted. + var limit = self.config.userMessageLimit + if quota.isSome() and quota.get().userMessageLimit < limit: + limit = quota.get().userMessageLimit + + if epochIndex != self.currentEpochIndex: + self.currentEpochIndex = epochIndex + self.sentInCurrentEpoch = 0 + + if self.sentInCurrentEpoch >= limit: + return err(RateLimitError.OverBudget) + + inc self.sentInCurrentEpoch return ok() - -proc dequeueReady*(self: RateLimitManager): seq[seq[byte]] = - ## Returns the set of queued messages that may be dispatched now - ## without exceeding the configured rate limit. - discard - -proc resetEpoch*(self: RateLimitManager) = - self.currentEpochStart = getTime() - self.sentInCurrentEpoch = 0 diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index f9f3229f6..fb7dfa1ee 100644 --- a/logos_delivery/waku/api/publish.nim +++ b/logos_delivery/waku/api/publish.nim @@ -8,7 +8,7 @@ import logos_delivery/waku/compat/option_valueor {.push raises: [].} -import std/options +import std/[options, times, strutils] import results, chronos import logos_delivery/waku/waku @@ -40,10 +40,59 @@ 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 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, some(shard)).isSome() diff --git a/logos_delivery/waku/node/waku_node/lightpush.nim b/logos_delivery/waku/node/waku_node/lightpush.nim index 56d39aae3..ab2c3ef63 100644 --- a/logos_delivery/waku/node/waku_node/lightpush.nim +++ b/logos_delivery/waku/node/waku_node/lightpush.nim @@ -105,30 +105,6 @@ proc resolveLegacyPubsubTopic( return err("Autosharding error: " & error) return ok($shard) -proc runRlnRefreshRetry( - node: WakuNode, - rln: Option[Rln], - msgWithProof: WakuMessage, - pubsubForPublish: PubsubTopic, - peer: RemotePeerInfo, - fallback: legacy_lightpush_protocol.WakuLightPushResult[string], -): Future[legacy_lightpush_protocol.WakuLightPushResult[string]] {.async, gcsafe.} = - ## Refreshes the RLN merkle proof path and retries the publish once. Only the - ## refresh is bounded by RlnMerkleProofRefreshTimeout (returning `fallback` on - ## timeout); the retried publish runs unbounded, matching the first attempt. - info "legacy lightpush send rejected as RLN-invalid; " & - "refreshing merkle proof and retrying once" - rln.get().groupManager.invalidateMerkleProofCache() - - let refreshFut = attachRLNProof(rln.get(), msgWithProof) - if not (await refreshFut.withTimeout(RlnMerkleProofRefreshTimeout)): - warn "legacy lightpush RLN proof refresh timed out; returning original error" - return fallback - let retryMsg = refreshFut.read().valueOr: - return err("failed call attachRLNProof from lightpush retry: " & error) - - return await internalLegacyLightpushPublish(node, pubsubForPublish, retryMsg, peer) - proc legacyLightpushPublish*( node: WakuNode, pubsubTopic: Option[PubsubTopic], @@ -161,18 +137,20 @@ proc legacyLightpushPublish*( ).valueOr: return err(error) - let firstResult = + let publishResult = await internalLegacyLightpushPublish(node, pubsubForPublish, msgWithProof, peer) # Legacy has no status codes, so string-match the RLN error to detect a - # stale merkle proof path, then refresh and retry once. - if firstResult.isOk() or rln.isNone() or - not firstResult.error.contains(RlnValidatorErrorMsg): - return firstResult + # stale merkle proof path. Schedule the refresh and hand the error back: + # retrying is the caller's decision, the same way the non-legacy path + # behaves. A retry regenerates the proof against the refreshed cache. + if publishResult.isOk() or rln.isNone() or + not publishResult.error.contains(RlnValidatorErrorMsg): + return publishResult - return await runRlnRefreshRetry( - node, rln, msgWithProof, pubsubForPublish, peer, firstResult - ) + info "legacy lightpush send rejected as RLN-invalid; scheduling merkle proof refresh" + rln.get().groupManager.scheduleMerkleProofRefresh() + return err(RlnProofRefreshScheduledMsg & ": " & publishResult.error) except CatchableError: return err(getCurrentExceptionMsg()) diff --git a/logos_delivery/waku/rln/constants.nim b/logos_delivery/waku/rln/constants.nim index 9f93d5fe0..518365b56 100644 --- a/logos_delivery/waku/rln/constants.nim +++ b/logos_delivery/waku/rln/constants.nim @@ -25,11 +25,6 @@ const RlnValidatorErrorMsg* = "RLN validation failed" const RlnProofRefreshScheduledMsg* = "stale RLN proof suspected; refresh scheduled, retry the publish" -# Bounds the legacy lightpush merkle proof refresh (eth_call refetch + proof -# regen) so a hanging RPC cannot stall the caller. The retried publish is not -# bounded. -const RlnMerkleProofRefreshTimeout* = 5.seconds - # inputs of the membership contract constructor # TODO may be able to make these constants private and put them inside the waku_rln_utils const diff --git a/logos_delivery/waku/rln/proof.nim b/logos_delivery/waku/rln/proof.nim index 8a29eb67b..8402f8cb1 100644 --- a/logos_delivery/waku/rln/proof.nim +++ b/logos_delivery/waku/rln/proof.nim @@ -54,23 +54,41 @@ proc toRLNSignal*(wakumessage: WakuMessage): seq[byte] = output = concat(wakumessage.payload, contentTopicBytes, @(timestampBytes)) return output -proc generateRLNProof*( - rln: Rln, input: seq[byte], senderEpochTime: float64 +proc generateRLNProofWithNonce( + rln: Rln, input: seq[byte], senderEpochTime: float64, nonce: Nonce ): Future[Result[seq[byte], string]] {.async.} = + ## Generates a proof against an already drawn `nonce`. Regenerating for an + ## unchanged (input, epoch, nonce) is safe: the revealed share is a function + ## of those three, so a regenerated proof reveals the same share and cannot + ## read as double-signalling. let epoch = rln.calcEpoch(senderEpochTime) - let nonce = rln.nonceManager.getNonce().valueOr: - return err("could not get new message id to generate an rln proof: " & $error) let proof = (await rln.groupManager.generateProof(input, epoch, nonce)).valueOr: return err("could not generate rln-v2 proof: " & $error) return ok(proof.encode().buffer) +proc generateRLNProof*( + rln: Rln, input: seq[byte], senderEpochTime: float64 +): Future[Result[seq[byte], string]] {.async.} = + let nonce = rln.nonceManager.getNonce().valueOr: + return err("could not get new message id to generate an rln proof: " & $error) + return await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce) + proc generateRLNProofWithRootRefresh*( rln: Rln, input: seq[byte], senderEpochTime: float64 ): Future[Result[seq[byte], string]] {.async.} = ## Generates an RLN proof and checks its merkle root against the ## acceptable-root window. If the root is stale, invalidates the cache and ## regenerates once against a refetched path. Returns the proof bytes. - let proofBytes = (await rln.generateRLNProof(input, senderEpochTime)).valueOr: + ## + ## The regeneration reuses the nonce drawn for the first attempt: only the + ## merkle path differs between the two, so drawing again would spend two + ## message ids from the epoch budget on a message that is sent once. That + ## would drift the budget the rate limit manager accounts for away from the + ## one the nonce manager enforces. + let nonce = rln.nonceManager.getNonce().valueOr: + return err("could not get new message id to generate an rln proof: " & $error) + + let proofBytes = (await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce)).valueOr: return err("failed to generate RLN proof: " & $error) let rlnProof = RateLimitProof.init(proofBytes).valueOr: @@ -81,7 +99,7 @@ proc generateRLNProofWithRootRefresh*( info "RLN: stale merkle root detected; refreshing merkle path and regenerating proof" rln.groupManager.invalidateMerkleProofCache() - return await rln.generateRLNProof(input, senderEpochTime) + return await rln.generateRLNProofWithNonce(input, senderEpochTime, nonce) proc attachRLNProof*( r: Rln, message: WakuMessage diff --git a/tests/messaging/test_all.nim b/tests/messaging/test_all.nim index 3cf508033..3608348e8 100644 --- a/tests/messaging/test_all.nim +++ b/tests/messaging/test_all.nim @@ -1,3 +1,3 @@ {.used.} -import ./test_rate_limit_manager +import ./test_rate_limit_manager, ./test_rln_proof_attach diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim index 895fcb49d..270a20137 100644 --- a/tests/messaging/test_rate_limit_manager.nim +++ b/tests/messaging/test_rate_limit_manager.nim @@ -1,26 +1,87 @@ {.used.} +import std/options import chronos, testutils/unittests, stew/byteutils import logos_delivery/messaging/rate_limit_manager/rate_limit_manager +proc fixedQuota(epochIndex, userMessageLimit: uint64): QuotaProvider = + ## A quota source pinned to one epoch — the epoch never rolls on its own, so + ## limit-boundary tests are deterministic without touching the wall clock. + return proc(): Option[EpochQuota] {.gcsafe, raises: [].} = + some(EpochQuota(epochIndex: epochIndex, userMessageLimit: userMessageLimit)) + suite "RateLimitManager - admission": asyncTest "admit is a pass-through when disabled": let rl = RateLimitManager.new( - RateLimitConfig(enabled: false, epochPeriodSec: 600, messagesPerEpoch: 1) + RateLimitConfig(enabled: false, epochSizeSec: 600, userMessageLimit: 1) ) for _ in 0 ..< 10: - let res = await rl.admit("payload".toBytes()) - check res.isOk() + check (await rl.admit("payload".toBytes())).isOk() - asyncTest "admit is a pass-through in the skeleton even when enabled": - ## Documents the current skeleton behaviour: per-epoch enforcement is - ## not wired yet, so every call is admitted regardless of the - ## configured budget. This test flips to red as soon as real - ## enforcement lands, at which point it should be replaced with - ## budget-boundary assertions. + asyncTest "zeroed limit or epoch size is treated as disabled": + let zeroLimit = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 0) + ) + check (await zeroLimit.admit("a".toBytes())).isOk() + + let zeroEpoch = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 0, userMessageLimit: 1) + ) + check (await zeroEpoch.admit("a".toBytes())).isOk() + check (await zeroEpoch.admit("b".toBytes())).isOk() + + asyncTest "admits up to the user message limit then rejects with OverBudget": + ## Fixed-epoch quota so the window cannot roll mid-test. let rl = RateLimitManager.new( - RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1) + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 3), + fixedQuota(epochIndex = 42, userMessageLimit = 100), + ) + for i in 0 ..< 3: + check (await rl.admit(("msg" & $i).toBytes())).isOk() + let res = await rl.admit("over".toBytes()) + check: + res.isErr() + res.error == RateLimitError.OverBudget + + asyncTest "allowance refills when the epoch rolls over": + ## Drive the roll through the provider — no sleeps, no flake. + var epoch = 1'u64 + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 1), + proc(): Option[EpochQuota] {.gcsafe, raises: [].} = + some(EpochQuota(epochIndex: epoch, userMessageLimit: 100)), ) check (await rl.admit("first".toBytes())).isOk() - check (await rl.admit("second".toBytes())).isOk() + check (await rl.admit("second".toBytes())).isErr() + epoch = 2 + check (await rl.admit("third".toBytes())).isOk() + check (await rl.admit("fourth".toBytes())).isErr() + + asyncTest "RLN user message limit clamps a looser configured limit": + ## config allows 5, RLN grants 2 — the third admission must reject, since + ## exceeding RLN's limit would fail at proof generation anyway. + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 5), + fixedQuota(epochIndex = 7, userMessageLimit = 2), + ) + check (await rl.admit("a".toBytes())).isOk() + check (await rl.admit("b".toBytes())).isOk() + check (await rl.admit("c".toBytes())).isErr() + + asyncTest "config can tighten below the RLN limit": + ## config allows 1, RLN grants 100 — the config limit wins. + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 1), + fixedQuota(epochIndex = 7, userMessageLimit = 100), + ) + check (await rl.admit("a".toBytes())).isOk() + check (await rl.admit("b".toBytes())).isErr() + + asyncTest "falls back to the wall-clock window when no quota source is set": + ## No provider: rate limiting still enforces within a single wall-clock epoch. + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochSizeSec: 600, userMessageLimit: 1) + ) + check (await rl.admit("first".toBytes())).isOk() + check (await rl.admit("second".toBytes())).isErr() diff --git a/tests/messaging/test_rln_proof_attach.nim b/tests/messaging/test_rln_proof_attach.nim new file mode 100644 index 000000000..a192969d8 --- /dev/null +++ b/tests/messaging/test_rln_proof_attach.nim @@ -0,0 +1,102 @@ +{.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 = 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 + +suite "SendService RLN proof attach - RLN mounted": + var + waku {.threadvar.}: Waku + anvilProc {.threadvar.}: Process + manager {.threadvar.}: OnchainGroupManager + + asyncSetup: + anvilProc = runAnvil(stateFile = 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 "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 diff --git a/tests/node/test_wakunode_legacy_lightpush.nim b/tests/node/test_wakunode_legacy_lightpush.nim index aec37e18c..69cc3e269 100644 --- a/tests/node/test_wakunode_legacy_lightpush.nim +++ b/tests/node/test_wakunode_legacy_lightpush.nim @@ -186,21 +186,24 @@ suite "RLN Proofs as a Lightpush Service": # The tests below drive `server.legacyLightpushPublish(...)` against the # server node. Because `server.wakuLegacyLightPush` is mounted (and no - # legacy client is), the call takes the self-request path — it still runs - # the full client-side flow (proof gen, retry on RlnValidatorErrorMsg - # substring, one-retry cap), but the request lands in the local + # legacy client is), the call takes the self-request path — it runs the + # full client-side flow (proof gen, RLN-rejection detection via the + # RlnValidatorErrorMsg substring), but the request lands in the local # pushHandler. Swapping in a stub pushHandler lets each test control what - # attempt N sees. + # the publish attempt sees. + # + # On an RLN rejection the publish schedules a background merkle-proof + # refresh and returns the error tagged with RlnProofRefreshScheduledMsg; + # the caller (the send service loop) regenerates the proof and republishes + # on its next round. - asyncTest "retry fires on RlnValidatorErrorMsg substring and second attempt succeeds": + asyncTest "RLN rejection schedules a refresh and surfaces the tagged error": var callCount = 0 let stub: PushMessageHandler = proc( pubsubTopic: PubsubTopic, message: WakuMessage ): Future[WakuLightPushResult[void]] {.async.} = inc callCount - if callCount == 1: - return err(RlnValidatorErrorMsg & ": simulated stale merkle path") - return ok() + return err(RlnValidatorErrorMsg & ": simulated stale merkle path") server.wakuLegacyLightPush.pushHandler = stub let response = await server.legacyLightpushPublish( @@ -208,8 +211,10 @@ suite "RLN Proofs as a Lightpush Service": ) check: - callCount == 2 - response.isOk() + callCount == 1 + response.isErr() + response.error.contains(RlnProofRefreshScheduledMsg) + response.error.contains(RlnValidatorErrorMsg) asyncTest "no retry when error does not contain RlnValidatorErrorMsg": var callCount = 0 @@ -229,27 +234,9 @@ suite "RLN Proofs as a Lightpush Service": response.isErr() response.error == "unrelated failure" - asyncTest "retry cap: two consecutive RLN errors surface the second": - var callCount = 0 - let stub: PushMessageHandler = proc( - pubsubTopic: PubsubTopic, message: WakuMessage - ): Future[WakuLightPushResult[void]] {.async.} = - inc callCount - return err(RlnValidatorErrorMsg & ": still stale") - server.wakuLegacyLightPush.pushHandler = stub - - let response = await server.legacyLightpushPublish( - some(pubsubTopic), message, server.peerInfo.toRemotePeerInfo() - ) - - check: - callCount == 2 - response.isErr() - response.error.contains(RlnValidatorErrorMsg) - - asyncTest "no retry when node.rln is nil": - # Detach RLN so the retry branch short-circuits on rln.isNone() even - # when the error string carries RlnValidatorErrorMsg. Restore before + asyncTest "no refresh scheduled when node.rln is nil": + # Detach RLN so the RLN-rejection branch short-circuits on rln.isNone() + # even when the error string carries RlnValidatorErrorMsg. Restore before # teardown so server.stop() sees the same object graph it was # constructed with. let savedRln = server.rln