mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-22 04:29:42 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
parent
5a9f518bdd
commit
ef06f67a2a
@ -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(
|
||||
|
||||
@ -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 =
|
||||
|
||||
@ -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*(
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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 —
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user