From 9e2a7815072f92af047d1a7775e8711ff9220153 Mon Sep 17 00:00:00 2001 From: stubbsta Date: Fri, 10 Jul 2026 14:34:54 +0200 Subject: [PATCH 1/4] feat: enforce per-epoch budget in RateLimitManager.admit Replaces the pass-through skeleton with a lazily rolled fixed window: admit() charges one message against the current epoch, resets the counter once epochPeriodSec has elapsed, and rejects with OverBudget when messagesPerEpoch is exhausted. Disabled or non-positive configurations admit everything, so the default-constructed MessagingClientConf (enabled = false) keeps today's behaviour. Parking of over-budget messages stays with the SendService scheduler (NextRoundRetry); the manager only answers whether one more transmission fits. The queue / dequeueReady stubs that anticipated manager-side parking are removed accordingly. Extends tests/messaging/test_rate_limit_manager.nim with budget boundary, epoch rollover, resetEpoch, and degenerate-config cases, replacing the enabled-pass-through placeholder test. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../rate_limit_manager/rate_limit_manager.nim | 48 +++++++++++-------- tests/messaging/test_rate_limit_manager.nim | 44 ++++++++++++++--- 2 files changed, 64 insertions(+), 28 deletions(-) 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..67aceae98 100644 --- a/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim @@ -1,12 +1,13 @@ ## 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. +## budget is exhausted, ensuring RLN compliance on enforcing relays. ## -## 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. +## Budgeting is a lazily rolled fixed window: the first admission after +## `epochPeriodSec` has elapsed resets the counter. Parking and retrying +## of over-budget messages is owned by the send service scheduler; this +## module only answers whether one more transmission fits the current +## epoch's budget. ## ## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html @@ -24,7 +25,6 @@ type RateLimitManager* = ref object config*: RateLimitConfig - queue*: seq[seq[byte]] currentEpochStart*: Time sentInCurrentEpoch*: int @@ -33,22 +33,28 @@ const DefaultMessagesPerEpoch* = 1 proc new*(T: type RateLimitManager, config: RateLimitConfig): T = - return - T(config: config, queue: @[], currentEpochStart: getTime(), sentInCurrentEpoch: 0) - -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`. - 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 + return T(config: config, currentEpochStart: getTime(), sentInCurrentEpoch: 0) proc resetEpoch*(self: RateLimitManager) = self.currentEpochStart = getTime() self.sentInCurrentEpoch = 0 + +proc admit*( + self: RateLimitManager, msg: seq[byte] +): Future[Result[void, RateLimitError]] {.async: (raises: []).} = + ## Charges one message against the current epoch's budget, rolling the + ## window first when `epochPeriodSec` has elapsed. A disabled or + ## non-positive configuration admits everything. + if not self.config.enabled or self.config.epochPeriodSec <= 0 or + self.config.messagesPerEpoch <= 0: + return ok() + + if getTime() - self.currentEpochStart >= + initDuration(seconds = self.config.epochPeriodSec): + self.resetEpoch() + + if self.sentInCurrentEpoch >= self.config.messagesPerEpoch: + return err(RateLimitError.OverBudget) + + inc self.sentInCurrentEpoch + return ok() diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim index 895fcb49d..bd259a19f 100644 --- a/tests/messaging/test_rate_limit_manager.nim +++ b/tests/messaging/test_rate_limit_manager.nim @@ -13,14 +13,44 @@ suite "RateLimitManager - admission": let res = await rl.admit("payload".toBytes()) check res.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 "admits up to the budget then rejects with OverBudget": + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 3) + ) + 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 "budget frees when the epoch rolls over": + let rl = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 1, messagesPerEpoch: 1) + ) + check (await rl.admit("first".toBytes())).isOk() + check (await rl.admit("second".toBytes())).isErr() + await sleepAsync(1100.milliseconds) + check (await rl.admit("third".toBytes())).isOk() + check (await rl.admit("fourth".toBytes())).isErr() + + asyncTest "resetEpoch forces a fresh budget": let rl = RateLimitManager.new( RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1) ) check (await rl.admit("first".toBytes())).isOk() - check (await rl.admit("second".toBytes())).isOk() + check (await rl.admit("second".toBytes())).isErr() + rl.resetEpoch() + check (await rl.admit("third".toBytes())).isOk() + + asyncTest "non-positive budget or period is treated as disabled": + let zeroBudget = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 0) + ) + check (await zeroBudget.admit("a".toBytes())).isOk() + + let zeroPeriod = RateLimitManager.new( + RateLimitConfig(enabled: true, epochPeriodSec: 0, messagesPerEpoch: 1) + ) + check (await zeroPeriod.admit("a".toBytes())).isOk() + check (await zeroPeriod.admit("b".toBytes())).isOk() From 1b9407cf9b200f791218038e3f9f43f11ebb6433 Mon Sep 17 00:00:00 2001 From: stubbsta Date: Tue, 14 Jul 2026 14:45:48 +0200 Subject: [PATCH 2/4] 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) --- .../send_service/send_service.nim | 18 ++++ logos_delivery/waku/api/publish.nim | 30 +++++- tests/messaging/test_all.nim | 2 +- tests/messaging/test_rln_proof_attach.nim | 102 ++++++++++++++++++ 4 files changed, 148 insertions(+), 4 deletions(-) create mode 100644 tests/messaging/test_rln_proof_attach.nim 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..a3a59bbaa 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim @@ -260,6 +260,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 +305,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/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index f9f3229f6..0c4b5ad03 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] import results, chronos import logos_delivery/waku/waku @@ -40,10 +40,34 @@ 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) + 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/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_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 From bed1fc6071b68f22c1fb7714de07a43fe7f25027 Mon Sep 17 00:00:00 2001 From: stubbsta Date: Wed, 15 Jul 2026 12:57:27 +0200 Subject: [PATCH 3/4] 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) --- .../send_service/lightpush_processor.nim | 11 +++++ .../send_service/relay_processor.nim | 15 ++++++ .../send_service/send_service.nim | 4 +- logos_delivery/waku/api/publish.nim | 27 +++++++++- .../waku/node/waku_node/lightpush.nim | 42 ++++------------ logos_delivery/waku/rln/constants.nim | 5 -- logos_delivery/waku/rln/proof.nim | 30 +++++++++--- tests/node/test_wakunode_legacy_lightpush.nim | 49 +++++++------------ 8 files changed, 107 insertions(+), 76 deletions(-) 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 a3a59bbaa..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)) diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index 0c4b5ad03..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, times] +import std/[options, times, strutils] import results, chronos import logos_delivery/waku/waku @@ -68,6 +68,31 @@ proc attachRlnProof*( 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/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 From 0674efff41cbf41a04be7832833639df6fcb607d Mon Sep 17 00:00:00 2001 From: stubbsta Date: Fri, 17 Jul 2026 13:15:54 +0200 Subject: [PATCH 4/4] feat: split rate limit manager into config, quota source, and enforcement modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Decomposes logos_delivery/messaging/rate_limit_manager/ into three modules with one responsibility each: - rate_limit_config: the configuration vocabulary (RateLimitConfig, RateLimitError, defaults, isEnforcing). The API conf layer now imports this alone instead of the enforcement engine. - quota_source: the RLN seam. A single QuotaProvider callback returns EpochQuota (epoch index + user message limit) so the two are read atomically and a read cannot straddle an epoch boundary. Returns none when RLN is unavailable, selecting the wall-clock fallback. The callback shape keeps the manager free of any dependency on the Waku kernel; the RLN-backed provider is built one layer up. - rate_limit_manager: enforcement only; re-exports the other two so existing single-import call sites are unchanged. Config field names now mirror RLN exactly, per the requirement that the rate limit share RLN Relay's format: epochSizeSec (was epochPeriodSec) and userMessageLimit (was messagesPerEpoch), both uint64 to match RlnConf. Renaming is contained to this branch: the config type moved into the new rate_limit_config module here. admit() now works in epoch-index terms: the epoch comes from the provider when set (RLN's calcEpoch value), else from an absolute wall-clock window (unixTime div epochSizeSec — absolute rather than anchored at first use, matching RLN's derivation). The effective limit is min(config.userMessageLimit, RLN's) — RLN can only tighten the configured limit, since exceeding it would fail at proof generation once the epoch's message ids are exhausted. With no provider wired (this commit), behaviour is the wall-clock window as before; RLN-backed provider wiring follows separately. Tests rewritten against injected fake providers: limit boundary, epoch rollover without sleeps, RLN-clamps-config, config-tightens-below-RLN, and the wall-clock fallback (7 tests). Co-Authored-By: Claude Opus 4.8 (1M context) --- logos_delivery/api/conf/messaging_conf.nim | 8 +- .../rate_limit_manager/quota_source.nim | 38 ++++++++ .../rate_limit_manager/rate_limit_config.nim | 31 ++++++ .../rate_limit_manager/rate_limit_manager.nim | 95 +++++++++++-------- tests/messaging/test_rate_limit_manager.nim | 79 ++++++++++----- 5 files changed, 186 insertions(+), 65 deletions(-) create mode 100644 logos_delivery/messaging/rate_limit_manager/quota_source.nim create mode 100644 logos_delivery/messaging/rate_limit_manager/rate_limit_config.nim 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/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 67aceae98..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,59 +1,80 @@ ## Rate Limit Manager for the Messaging API. ## -## Tracks messages sent per RLN epoch and rejects admission when the -## budget is exhausted, 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. ## -## Budgeting is a lazily rolled fixed window: the first admission after -## `epochPeriodSec` has elapsed resets the counter. Parking and retrying -## of over-budget messages is owned by the send service scheduler; this -## module only answers whether one more transmission fits the current -## epoch's budget. +## 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 - 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, currentEpochStart: getTime(), sentInCurrentEpoch: 0) - -proc resetEpoch*(self: RateLimitManager) = - self.currentEpochStart = getTime() - self.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: []).} = - ## Charges one message against the current epoch's budget, rolling the - ## window first when `epochPeriodSec` has elapsed. A disabled or - ## non-positive configuration admits everything. - if not self.config.enabled or self.config.epochPeriodSec <= 0 or - self.config.messagesPerEpoch <= 0: + ## 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() - if getTime() - self.currentEpochStart >= - initDuration(seconds = self.config.epochPeriodSec): - self.resetEpoch() + let quota = self.currentQuota() - if self.sentInCurrentEpoch >= self.config.messagesPerEpoch: + 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 diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim index bd259a19f..270a20137 100644 --- a/tests/messaging/test_rate_limit_manager.nim +++ b/tests/messaging/test_rate_limit_manager.nim @@ -1,21 +1,41 @@ {.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 "admits up to the budget then rejects with OverBudget": + 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: 3) + 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() @@ -24,33 +44,44 @@ suite "RateLimitManager - admission": res.isErr() res.error == RateLimitError.OverBudget - asyncTest "budget frees when the epoch rolls over": + 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, epochPeriodSec: 1, messagesPerEpoch: 1) + 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())).isErr() - await sleepAsync(1100.milliseconds) + epoch = 2 check (await rl.admit("third".toBytes())).isOk() check (await rl.admit("fourth".toBytes())).isErr() - asyncTest "resetEpoch forces a fresh budget": + 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, epochPeriodSec: 600, messagesPerEpoch: 1) + 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() - rl.resetEpoch() - check (await rl.admit("third".toBytes())).isOk() - - asyncTest "non-positive budget or period is treated as disabled": - let zeroBudget = RateLimitManager.new( - RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 0) - ) - check (await zeroBudget.admit("a".toBytes())).isOk() - - let zeroPeriod = RateLimitManager.new( - RateLimitConfig(enabled: true, epochPeriodSec: 0, messagesPerEpoch: 1) - ) - check (await zeroPeriod.admit("a".toBytes())).isOk() - check (await zeroPeriod.admit("b".toBytes())).isOk()