From 99854601b486dca4ae377a1ff024225ba6c1dbbf Mon Sep 17 00:00:00 2001 From: stubbsta Date: Wed, 8 Jul 2026 13:59:37 +0200 Subject: [PATCH] chore: relocate RateLimitManager to messaging layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moves rate_limit_manager.nim from `logos_delivery/channels/` to `logos_delivery/messaging/rate_limit_manager/` — the correct owner now that admission sits alongside RLN in the messaging layer instead of being fanned out to per-channel event listeners. The API surface changes: - `enqueueToSend` + `ReadyToSendEvent` (broker-based fan-out to a ReliableChannel listener) is replaced by `admit(msg): Future[Result[ void, RateLimitError]]`. Callers now branch directly on the result instead of subscribing to an event. - `channelId`, `SdsChannelID` and the SDS import are dropped — the messaging layer has no notion of channels; SDS was a channels-layer concern that only survived on this type because of the old broker fan-out. - `brokerCtx` is dropped for the same reason. Epoch config (`epochPeriodSec`), the wall-clock `currentEpochStart`, `queue`, `dequeueReady`, and `resetEpoch` are preserved exactly as the original owner designed them — this refactor is intentionally scoped to the API surface, not the epoch mechanism. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../rate_limit_manager/rate_limit_manager.nim | 80 ------------------- .../rate_limit_manager/rate_limit_manager.nim | 54 +++++++++++++ 2 files changed, 54 insertions(+), 80 deletions(-) delete mode 100644 logos_delivery/channels/rate_limit_manager/rate_limit_manager.nim create mode 100644 logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim diff --git a/logos_delivery/channels/rate_limit_manager/rate_limit_manager.nim b/logos_delivery/channels/rate_limit_manager/rate_limit_manager.nim deleted file mode 100644 index ab5a9f67b..000000000 --- a/logos_delivery/channels/rate_limit_manager/rate_limit_manager.nim +++ /dev/null @@ -1,80 +0,0 @@ -## Rate Limit Manager for the Reliable Channel API. -## -## Tracks messages sent per RLN epoch and delays dispatch when the -## limit is approached, ensuring RLN compliance on enforcing relays. -## -## For the skeleton this is a pass-through: messages are immediately -## released as ready-to-send. Real epoch budgeting will be added later. -## -## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html - -import std/times -import message -import brokers/event_broker -import brokers/broker_context - -export event_broker, broker_context -export message.SdsChannelID - -const - DefaultEpochPeriodSec* = 600 - DefaultMessagesPerEpoch* = 1 - -EventBroker: - ## Emitted by `enqueueToSend` carrying the batch of opaque message - ## blobs that may now leave the rate limiter and continue down the - ## outgoing pipeline (encryption -> dispatch). Bytes only: the rate - ## limiter is intentionally agnostic of SDS, so anything serialisable - ## can flow through it. - ## - ## `channelId` lets listeners filter to their own channel, since all - ## reliable channels share the underlying Waku node's broker context. - type ReadyToSendEvent* = ref object - channelId*: SdsChannelID - msgs*: seq[seq[byte]] - -type - RateLimitConfig* = object - enabled*: bool ## spec: rate limiting opt-in; SHOULD be true when RLN active - epochPeriodSec*: int - messagesPerEpoch*: int - - RateLimitManager* = ref object - config*: RateLimitConfig - queue*: seq[seq[byte]] - currentEpochStart*: Time - sentInCurrentEpoch*: int - channelId*: SdsChannelID ## tag for the emitted `ReadyToSendEvent` - brokerCtx: BrokerContext - -proc new*( - T: type RateLimitManager, - config: RateLimitConfig, - channelId: SdsChannelID, - brokerCtx: BrokerContext = globalBrokerContext(), -): T = - return T( - config: config, - queue: @[], - currentEpochStart: getTime(), - sentInCurrentEpoch: 0, - channelId: channelId, - brokerCtx: brokerCtx, - ) - -proc enqueueToSend*(self: RateLimitManager, msg: seq[byte]) = - ## Skeleton behaviour: enqueue and immediately release as a single - ## ready batch. Real per-epoch budgeting will park messages on - ## `self.queue` and emit only when the budget allows. - ReadyToSendEvent.emit( - self.brokerCtx, ReadyToSendEvent(channelId: self.channelId, msgs: @[msg]) - ) - -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/messaging/rate_limit_manager/rate_limit_manager.nim b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim new file mode 100644 index 000000000..7ee197d8a --- /dev/null +++ b/logos_delivery/messaging/rate_limit_manager/rate_limit_manager.nim @@ -0,0 +1,54 @@ +## 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. +## +## 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. +## +## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html + +import std/times +import results, chronos + +type + RateLimitError* {.pure.} = enum + OverBudget + + RateLimitConfig* = object + enabled*: bool ## spec: rate limiting opt-in; SHOULD be true when RLN active + epochPeriodSec*: int + messagesPerEpoch*: int + + RateLimitManager* = ref object + config*: RateLimitConfig + queue*: seq[seq[byte]] + currentEpochStart*: Time + sentInCurrentEpoch*: int + +const + DefaultEpochPeriodSec* = 600 + 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 + +proc resetEpoch*(self: RateLimitManager) = + self.currentEpochStart = getTime() + self.sentInCurrentEpoch = 0