From ef06f67a2a47ff27789c233ff04a933c50ba60da Mon Sep 17 00:00:00 2001 From: stubbsta Date: Tue, 21 Jul 2026 13:40:00 +0200 Subject: [PATCH] feat: source the rate limit manager's epoch and budget from RLN Wires the quota seam to its producer, so enforcement tracks RLN rather than only the wall clock. - Waku.currentRlnEpochQuota (waku/api/publish) reads RLN's current epoch index and the epoch's user message limit together, returning none when RLN is not mounted (or its limit is unset). - MessagingClient.new builds a QuotaProvider closure over that accessor and hands it to the RateLimitManager. The closure is late-binding: it queries the kernel on each admission, so a node whose RLN mounts after construction upgrades from the wall-clock fallback to RLN's epoch and limit automatically, with no reconstruction. With this, admit() rolls its window on RLN's epoch and clamps the configured cap to RLN's user message limit; without RLN it still falls back to the absolute wall-clock window and the configured limit. Also switches the quota seam from std/options to results `Opt`, matching the kernel surface it now bridges (`groupManager.userMessageLimit` is `Opt`) and the rest of the messaging layer post-Opt migration. Tests: currentRlnEpochQuota is none unmounted and reports epoch + the configured userMessageLimit when mounted (anvil-backed). Co-Authored-By: Claude Opus 4.8 (1M context) --- logos_delivery/messaging/messaging_client.nim | 17 ++++++++++++++++- .../rate_limit_manager/quota_source.nim | 5 +++-- .../rate_limit_manager/rate_limit_manager.nim | 5 ++--- logos_delivery/waku/api/publish.nim | 13 +++++++++++++ tests/messaging/test_rate_limit_manager.nim | 11 +++++------ tests/messaging/test_rln_proof_attach.nim | 14 ++++++++++++++ 6 files changed, 53 insertions(+), 12 deletions(-) diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim index c33b3ff6a..3adb85c90 100644 --- a/logos_delivery/messaging/messaging_client.nim +++ b/logos_delivery/messaging/messaging_client.nim @@ -6,6 +6,7 @@ import logos_delivery/api/conf/messaging_conf, logos_delivery/api/messaging_client_api, logos_delivery/waku/waku, + logos_delivery/waku/api/publish, logos_delivery/waku/factory/conf_builder/waku_conf_builder, logos_delivery/messaging/delivery_service/[recv_service, send_service], logos_delivery/messaging/rate_limit_manager/rate_limit_manager @@ -19,6 +20,16 @@ type MessagingClient* = ref object recvService*: RecvService started*: bool +proc rlnQuotaProvider(waku: Waku): QuotaProvider = + ## Sources the rate limit manager's epoch + limit from RLN. Late-binding: the + ## closure queries `waku` on each admission, so a node whose RLN mounts after + ## this client is constructed upgrades from the wall-clock fallback to RLN's + ## epoch and user message limit automatically. + return proc(): Opt[EpochQuota] {.gcsafe, raises: [].} = + let q = waku.currentRlnEpochQuota().valueOr: + return Opt.none(EpochQuota) + Opt.some(EpochQuota(epochIndex: q.epochIndex, userMessageLimit: q.messageLimit)) + proc new*( T: type MessagingClient, conf: MessagingClientConf, waku: Waku ): Result[T, string] = @@ -26,7 +37,11 @@ proc new*( ## for transport while exposing its own send/recv API. let reliability = conf.reliabilityEnabled.get(DefaultP2pReliability) let sendService = ?SendService.new( - reliability, waku, RateLimitManager.new(conf.rateLimit.get(DefaultRateLimitConfig)) + reliability, + waku, + RateLimitManager.new( + conf.rateLimit.get(DefaultRateLimitConfig), rlnQuotaProvider(waku) + ), ) let recvService = RecvService.new(waku) return ok( diff --git a/logos_delivery/messaging/rate_limit_manager/quota_source.nim b/logos_delivery/messaging/rate_limit_manager/quota_source.nim index f46132a43..4fb8ac064 100644 --- a/logos_delivery/messaging/rate_limit_manager/quota_source.nim +++ b/logos_delivery/messaging/rate_limit_manager/quota_source.nim @@ -15,7 +15,8 @@ ## scope. It returns `none` when RLN is not mounted, which is the signal to ## fall back to the wall clock. -import std/[options, times] +import std/times +import results type EpochQuota* = object @@ -26,7 +27,7 @@ type ## Message ids the epoch grants — the hard ceiling on admissions before ## the epoch rolls over. - QuotaProvider* = proc(): Option[EpochQuota] {.gcsafe, raises: [].} + QuotaProvider* = proc(): Opt[EpochQuota] {.gcsafe, raises: [].} ## The epoch and its user message limit, or `none` when RLN is not mounted. proc wallClockEpochIndex*(epochPeriodSec: uint64): uint64 = 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 d6fb14724..b115f5143 100644 --- a/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim @@ -13,7 +13,6 @@ ## ## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html -import std/options import results, chronos import ./rate_limit_config, ./quota_source @@ -41,9 +40,9 @@ proc new*( sentInCurrentEpoch: 0, ) -proc currentQuota(self: RateLimitManager): Option[EpochQuota] = +proc currentQuota(self: RateLimitManager): Opt[EpochQuota] = if self.quotaProvider.isNil(): - return none(EpochQuota) + return Opt.none(EpochQuota) return self.quotaProvider() proc admit*( diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim index 98372c9a5..ce8973be1 100644 --- a/logos_delivery/waku/api/publish.nim +++ b/logos_delivery/waku/api/publish.nim @@ -82,6 +82,19 @@ func isRlnRejection*(error: ErrorStatus): bool = error.desc.get("").contains(RlnValidatorErrorMsg) ) +proc currentRlnEpochQuota*(self: Waku): Opt[tuple[epochIndex, messageLimit: uint64]] = + ## RLN's current epoch index and the epoch's user message limit, read + ## together so the pair cannot straddle an epoch boundary. `none` when RLN is + ## not mounted (or its limit is unset) — which the rate limit manager reads as + ## "fall back to the wall-clock window and the configured limit". + if self.node.rln.isNil(): + return Opt.none(tuple[epochIndex, messageLimit: uint64]) + + let limit = self.node.rln.groupManager.userMessageLimit.valueOr: + return Opt.none(tuple[epochIndex, messageLimit: uint64]) + + return Opt.some((fromEpoch(self.node.rln.getCurrentEpoch()), uint64(limit))) + 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 diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim index 68908d482..9a3ec6788 100644 --- a/tests/messaging/test_rate_limit_manager.nim +++ b/tests/messaging/test_rate_limit_manager.nim @@ -1,15 +1,14 @@ {.used.} -import std/options -import chronos, testutils/unittests, stew/byteutils +import results, 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)) + return proc(): Opt[EpochQuota] {.gcsafe, raises: [].} = + Opt.some(EpochQuota(epochIndex: epochIndex, userMessageLimit: userMessageLimit)) suite "RateLimitManager - admission": asyncTest "admit is a pass-through when disabled": @@ -49,8 +48,8 @@ suite "RateLimitManager - admission": var epoch = 1'u64 let rl = RateLimitManager.new( RateLimitConfig(enabled: true, epochPeriodSec: 600, messagesPerEpoch: 1), - proc(): Option[EpochQuota] {.gcsafe, raises: [].} = - some(EpochQuota(epochIndex: epoch, userMessageLimit: 100)), + proc(): Opt[EpochQuota] {.gcsafe, raises: [].} = + Opt.some(EpochQuota(epochIndex: epoch, userMessageLimit: 100)), ) 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 index 10c101e3f..11f7d754a 100644 --- a/tests/messaging/test_rln_proof_attach.nim +++ b/tests/messaging/test_rln_proof_attach.nim @@ -48,6 +48,11 @@ suite "SendService RLN proof attach": 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 @@ -90,6 +95,15 @@ suite "SendService RLN proof attach - RLN mounted": 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 —