From 9f3e7d0ee6431dfd2fd23a4f0741e8f268464c51 Mon Sep 17 00:00:00 2001 From: stubbsta Date: Wed, 8 Jul 2026 13:43:53 +0200 Subject: [PATCH 1/3] chore: drop rate-limit stage from reliable channel send pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Collapses the outgoing pipeline to `segmentation -> sds -> encryption -> dispatch` by folding the encrypt-and-dispatch tail of `onReadyToSend` directly into `send()`. Removes the `RateLimitManager` field, its constructor param, the `ReadyToSendEvent` listener, and the `awaitingDispatch` accounting that only existed to bridge the event-bus hop between `send()` and `onReadyToSend`. The `rate_limit_manager.nim` module itself is untouched — it will be relocated to the messaging layer (co-located with RLN) in a follow-up commit, where per-epoch admission actually belongs. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../channels/api/channel_lifecycle.nim | 10 - logos_delivery/channels/api/send.nim | 2 +- logos_delivery/channels/reliable_channel.nim | 191 +++++------------- .../test_reliable_channel_send_receive.nim | 17 +- 4 files changed, 62 insertions(+), 158 deletions(-) diff --git a/logos_delivery/channels/api/channel_lifecycle.nim b/logos_delivery/channels/api/channel_lifecycle.nim index 0d32e2acf..57d4e83fe 100644 --- a/logos_delivery/channels/api/channel_lifecycle.nim +++ b/logos_delivery/channels/api/channel_lifecycle.nim @@ -51,15 +51,6 @@ proc createReliableChannel*( causalHistorySize: cc.sdsCausalHistorySize.get(DefaultCausalHistorySize), persistence: sdsPersistence(), ) - let rateConfig = RateLimitConfig( - # Setting a rate-limit parameter implies enabling; an explicit - # rateLimitEnabled still wins. - enabled: cc.rateLimitEnabled.get( - cc.rateLimitEpochPeriodSec.isSome() or cc.rateLimitMessagesPerEpoch.isSome() - ), - epochPeriodSec: cc.rateLimitEpochPeriodSec.get(DefaultEpochPeriodSec), - messagesPerEpoch: cc.rateLimitMessagesPerEpoch.get(DefaultMessagesPerEpoch), - ) let chn = ReliableChannel.new( channelId = channelId, @@ -67,7 +58,6 @@ proc createReliableChannel*( senderId = senderId, segConfig = segConfig, sdsConfig = sdsConfig, - rateConfig = rateConfig, brokerCtx = self.brokerCtx, ) diff --git a/logos_delivery/channels/api/send.nim b/logos_delivery/channels/api/send.nim index 23c41ac13..7af5fdca0 100644 --- a/logos_delivery/channels/api/send.nim +++ b/logos_delivery/channels/api/send.nim @@ -14,7 +14,7 @@ proc send*( ): Future[Result[RequestId, string]] {.async: (raises: []).} = ## Spec-level entry point. Looks the channel up by id and delegates ## to `ReliableChannel.send`, which exposes the visible pipeline - ## segmentation -> sds -> rate_limit_manager -> encryption. + ## segmentation -> sds -> encryption -> dispatch. let chn = self.channels.getOrDefault(channelId) if chn.isNil(): return err("unknown channel: " & channelId) diff --git a/logos_delivery/channels/reliable_channel.nim b/logos_delivery/channels/reliable_channel.nim index ca74f9df4..0286e8f27 100644 --- a/logos_delivery/channels/reliable_channel.nim +++ b/logos_delivery/channels/reliable_channel.nim @@ -1,10 +1,10 @@ ## Reliable Channel type. ## ## A `ReliableChannel` orchestrates segmentation, SDS (end-to-end -## reliability), optional encryption, and rate-limited dispatch on top -## of the Messaging API for a single channel. +## reliability), optional encryption, and dispatch on top of the +## Messaging API for a single channel. ## -## Outgoing pipeline: Segment -> SDS -> Rate Limit -> Encrypt -> Dispatch +## Outgoing pipeline: Segment -> SDS -> Encrypt -> Dispatch ## Incoming pipeline: Decrypt -> SDS -> Reassemble -> Emit event ## ## Channels are owned by a `ReliableChannelManager`. Lifecycle and send @@ -29,12 +29,9 @@ import logos_delivery/waku/waku_core/topics import ./segmentation/segmentation import ./scalable_data_sync/scalable_data_sync -import ./rate_limit_manager/rate_limit_manager import ./encryption/encryption -export - types, reliable_channel_manager_api, segmentation, scalable_data_sync, - rate_limit_manager, encryption +export types, reliable_channel_manager_api, segmentation, scalable_data_sync, encryption const LipWireReliableChannelVersion* = "RELIABLE-CHANNEL-API/1" ## Wire-format spec marker for the Reliable Channel layer, as defined @@ -51,33 +48,22 @@ type ChannelReqState = object ## Per channel-level request, tracks how many of its segments are - ## still queued, in flight, or have terminated. The channel-level - ## final event fires when `confirmedCount + failedCount` reaches - ## `totalExpectedSegments` AND no segments are still awaiting dispatch - ## or in flight. + ## still in flight or have terminated. The channel-level final event + ## fires when `confirmedCount + failedCount` reaches + ## `totalExpectedSegments` AND no segments are still in flight. persistenceReqType: MessagePersistence totalExpectedSegments: int ## Total segments produced by `segmentation.performSegmentation` ## for this `channelReqId`. Set once in `send`, never mutated. - awaitingDispatch: int - ## Segments enqueued in `rate_limit_manager` but not yet claimed - ## by `onReadyToSend`. Decremented when `onReadyToSend` picks a - ## message and assigns it to this `channelReqId`. inflightMessagingIds: seq[RequestId] ## Messaging-layer ids minted by the send handler that have not ## yet produced a final event. Removed on `MessageSentEvent` / `MessageErrorEvent`. confirmedCount: int failedCount: int - ChannelReqs = OrderedTable[RequestId, ChannelReqState] + ChannelReqs = Table[RequestId, ChannelReqState] ## Key: channelReqId (the parent id returned by channel `send`). Value: ## per-request state, see `ChannelReqState`. - ## - ## `OrderedTable` preserves insertion order, which matches the FIFO - ## order `rate_limit_manager` re-emits messages in: `onReadyToSend` - ## routes each segment to the first entry with `awaitingDispatch > 0`, - ## and that scan is correct precisely because the outer iteration - ## order matches the order `send` pushed entries. ReliableChannel* = ref object ## Spec-defined public type. Fields are private so callers cannot @@ -89,7 +75,6 @@ type rng: libp2p_crypto.Rng segmentation: SegmentationHandler sdsHandler: SdsHandler - rateLimit: RateLimitManager channelReqs: ChannelReqs brokerCtx: BrokerContext @@ -102,7 +87,6 @@ func init( return ChannelReqState( persistenceReqType: persistenceReqType, totalExpectedSegments: totalExpectedSegments, - awaitingDispatch: totalExpectedSegments, inflightMessagingIds: @[], confirmedCount: 0, failedCount: 0, @@ -123,8 +107,8 @@ proc stop*(self: ReliableChannel) {.async: (raises: []).} = proc tryFinalizeChannelReq(self: ReliableChannel, channelReqId: RequestId) = ## Tries to finalize the channel-level request identified by `channelReqId` if - ## certain conditions are met, i.e., no segments are still awaiting dispatch or in flight, - ## and the total number of confirmed + failed segments equals the total expected segments. + ## certain conditions are met, i.e., no segments are still in flight and the + ## total number of confirmed + failed segments equals the total expected segments. ## Therefore, the channel-level request is removed from `self.channelReqs` ## and the appropriate final event is emitted. ## @@ -132,7 +116,7 @@ proc tryFinalizeChannelReq(self: ReliableChannel, channelReqId: RequestId) = if state.totalExpectedSegments == 0: ## Either already finalized (and removed) or never inserted. return - if state.awaitingDispatch != 0 or state.inflightMessagingIds.len != 0: + if state.inflightMessagingIds.len != 0: return if state.confirmedCount + state.failedCount < state.totalExpectedSegments: return @@ -154,22 +138,6 @@ proc tryFinalizeChannelReq(self: ReliableChannel, channelReqId: RequestId) = ChannelMessageSentEvent(channelId: self.channelId, requestId: channelReqId), ) -type ClaimedSegment = object - channelReqId: RequestId - isEphemeral: bool - -proc claimAwaitingChannelReq(self: ReliableChannel): Opt[ClaimedSegment] = - for channelReqId, state in self.channelReqs.mpairs: - if state.awaitingDispatch > 0: - state.awaitingDispatch.dec() - return Opt.some( - ClaimedSegment( - channelReqId: channelReqId, - isEphemeral: state.persistenceReqType == MessagePersistence.Ephemeral, - ) - ) - return Opt.none(ClaimedSegment) - type MessagingOutcome {.pure.} = enum Sent Failed @@ -208,62 +176,61 @@ proc markSegmentInflight( error "unreachable: channelReqId not found in markSegmentInflight", channelReqId = $channelReqId, error = e.msg -proc onReadyToSend( - self: ReliableChannel, readyToSendEvent: ReadyToSendEvent -) {.async: (raises: []).} = - ## Tail of the outgoing pipeline. Invoked from the `ReadyToSendEvent` - ## listener once `rate_limit_manager` releases a batch of opaque - ## blobs (already-encoded SDS messages): +proc send*( + self: ReliableChannel, payload: seq[byte], ephemeral: bool = false +): Future[Result[RequestId, string]] {.async: (raises: []).} = + ## Single application-level send: ## - ## ... -> rate_limit_manager -> [encryption] -> dispatch + ## segmentation -> sds -> encryption -> dispatch ## - ## For each `m`, the next channelReqId still queued in rate-limit - ## claims the slot (FIFO across sibling sends). The channelReqId is - ## captured up front and used as a stable key for every later state - ## update — no positional index is ever held across an `await`, so - ## sibling events mutating other entries (or even this one's - ## `inflightMessagingIds`) cannot corrupt this fiber's view. - for m in readyToSendEvent.msgs: - let claimed = self.claimAwaitingChannelReq().valueOr: - ## rate_limit_manager emitted more messages than we have pending — - ## should not happen given `send` increments `awaitingDispatch` - ## once per enqueued SDS payload. Drop silently rather than - ## corrupt state. - break - let channelReqId = claimed.channelReqId - let isEphemeral = claimed.isEphemeral + ## The returned `RequestId` is the channel-level parent of one-or-more + ## messaging-layer `RequestId`s; the mapping is held in + ## `self.channelReqs` until every segment is final. + if payload.len == 0: + return err("empty payload") + let channelReqId = RequestId.new(self.rng) + let persistenceReqType = + if ephemeral: MessagePersistence.Ephemeral else: MessagePersistence.Persistent + + var sdsSegments: seq[seq[byte]] + for segmentBytes in self.segmentation.performSegmentation(payload): + ## Segments arrive already encoded; the segmentation module owns + ## the wire format so SDS only ever sees opaque bytes. + let sdsBytes = (await self.sdsHandler.wrapOutgoing(segmentBytes)).valueOr: + return err("SDS wrap failed: " & error) + sdsSegments.add(sdsBytes) + + self.channelReqs[channelReqId] = + ChannelReqState.init(persistenceReqType, sdsSegments.len) + + for sdsBytes in sdsSegments: ## TODO: revisit which fields of the SDS message must be encrypted. ## Encrypting the whole encoded blob forces every receiver to attempt ## decryption before it can route, which breaks selective dispatch. ## Leave routing metadata (channelId, causal-history references) in ## clear and encrypt only the application payload. - let encRes = await Encrypt.request(m) - let encrypted = encRes.valueOr: + let encrypted = (await Encrypt.request(sdsBytes)).valueOr: MessageErrorEvent.emit( self.brokerCtx, MessageErrorEvent( requestId: channelReqId, messageHash: "", error: "encryption failed: " & error ), ) - ## Encryption failed *before* we could hand the segment to the self.markSegmentFailed(channelReqId) continue - let wireBytes = seq[byte](encrypted) ## The `meta` field carries the Reliable Channel wire-format spec ## marker so the ingress side of any peer can route this WakuMessage ## to its Reliable Channel layer. let envelope = MessageEnvelope( contentTopic: self.contentTopic, - payload: wireBytes, - ephemeral: isEphemeral, + payload: seq[byte](encrypted), + ephemeral: ephemeral, meta: LipWireReliableChannelVersion.toBytes(), ) - let sendRes = await MessagingSend.request(self.brokerCtx, envelope) - - let messagingReqId = sendRes.valueOr: + let messagingReqId = (await MessagingSend.request(self.brokerCtx, envelope)).valueOr: MessageErrorEvent.emit( self.brokerCtx, MessageErrorEvent( @@ -277,45 +244,6 @@ proc onReadyToSend( self.markSegmentInflight(channelReqId, messagingReqId) -proc send*( - self: ReliableChannel, payload: seq[byte], ephemeral: bool = false -): Future[Result[RequestId, string]] {.async: (raises: []).} = - ## Single application-level send. The first three stages of the - ## outgoing pipeline are chained explicitly so the flow is visible - ## at a glance: - ## - ## segmentation -> sds -> rate_limit_manager - ## - ## `rate_limit_manager.enqueueToSend` emits a `ReadyToSendEvent` with - ## the SDS messages cleared for transmission; the channel's listener - ## then runs the final stage (encryption -> dispatch). - ## - ## The returned `RequestId` is the channel-level parent of one-or-more - ## messaging-layer `RequestId`s; the mapping is held in - ## `self.channelReqs` until every segment is final. - if payload.len == 0: - return err("empty payload") - - let channelReqId = RequestId.new(self.rng) - let persistenceReqType = - if ephemeral: MessagePersistence.Ephemeral else: MessagePersistence.Persistent - - var segmentCount = 0 - var enqueued: seq[seq[byte]] - for segmentBytes in self.segmentation.performSegmentation(payload): - ## Segments arrive already encoded; the segmentation module owns - ## the wire format so SDS only ever sees opaque bytes. - let sdsBytes = (await self.sdsHandler.wrapOutgoing(segmentBytes)).valueOr: - return err("SDS wrap failed: " & error) - enqueued.add(sdsBytes) - segmentCount.inc() - - self.channelReqs[channelReqId] = - ChannelReqState.init(persistenceReqType, segmentCount) - - for sdsBytes in enqueued: - self.rateLimit.enqueueToSend(sdsBytes) - return ok(channelReqId) proc reportReceived(self: ReliableChannel, content: seq[byte]) = @@ -335,8 +263,7 @@ proc reportReceived(self: ReliableChannel, content: seq[byte]) = ) proc dispatchRepair(self: ReliableChannel, wire: seq[byte]) {.async: (raises: []).} = - ## Repair rebroadcasts skip the rate-limit queue — its emissions are - ## claimed FIFO by pending sends. Pacing is done by SDS itself. + ## SDS-driven repair rebroadcast. Pacing is done by SDS itself. let encRes = await Encrypt.request(wire) let encrypted = encRes.valueOr: debug "SDS repair rebroadcast dropped: encryption failed", @@ -406,15 +333,13 @@ proc new*( senderId: SdsParticipantID, segConfig: SegmentationConfig, sdsConfig: SdsConfig, - rateConfig: RateLimitConfig, brokerCtx: BrokerContext = globalBrokerContext(), ): T = - ## Pipeline handlers (segmentation/SDS/rate-limit) are constructed - ## inside the channel rather than handed in by the caller — they are - ## implementation details of the channel, not knobs the API consumer - ## should be wiring up. Encryption is delegated to the `Encrypt`/ - ## `Decrypt` request brokers, so the channel keeps no per-instance - ## encryption state either. + ## Pipeline handlers (segmentation/SDS) are constructed inside the + ## channel rather than handed in by the caller — they are implementation + ## details of the channel, not knobs the API consumer should be wiring + ## up. Encryption is delegated to the `Encrypt`/`Decrypt` request + ## brokers, so the channel keeps no per-instance encryption state either. let chn = T( channelId: channelId, contentTopic: contentTopic, @@ -422,8 +347,7 @@ proc new*( rng: libp2p_crypto.newRng(), segmentation: SegmentationHandler.new(segConfig), sdsHandler: SdsHandler.new(sdsConfig, channelId, senderId), - rateLimit: RateLimitManager.new(rateConfig, channelId, brokerCtx), - channelReqs: initOrderedTable[RequestId, ChannelReqState](), + channelReqs: initTable[RequestId, ChannelReqState](), brokerCtx: brokerCtx, ) @@ -432,20 +356,11 @@ proc new*( asyncSpawn chn.dispatchRepair(wire) chn.sdsHandler.start() - ## Each channel owns its own egress + ingress + send-completion - ## listeners on `chn.brokerCtx`, filtered to traffic addressed to - ## this channel. Keeping the listeners (and the handler procs they - ## call) inside the channel lets `onReadyToSend` / - ## `onMessageReceived` / `onMessageFinal` stay private — the - ## manager doesn't need to know about them. - discard ReadyToSendEvent.listen( - chn.brokerCtx, - proc(evt: ReadyToSendEvent): Future[void] {.async: (raises: []).} = - if evt.channelId == chn.channelId: - await chn.onReadyToSend(evt) - , - ) - + ## Each channel owns its own ingress + send-completion listeners on + ## `chn.brokerCtx`, filtered to traffic addressed to this channel. + ## Keeping the listeners (and the handler procs they call) inside the + ## channel lets `onMessageReceived` / `onMessageFinal` stay private — + ## the manager doesn't need to know about them. discard MessageReceivedEvent.listen( chn.brokerCtx, proc(evt: MessageReceivedEvent): Future[void] {.async: (raises: []).} = diff --git a/tests/channels/test_reliable_channel_send_receive.nim b/tests/channels/test_reliable_channel_send_receive.nim index 53a88f27a..8a7b02231 100644 --- a/tests/channels/test_reliable_channel_send_receive.nim +++ b/tests/channels/test_reliable_channel_send_receive.nim @@ -336,9 +336,9 @@ suite "Reliable Channel - send state machine": asyncTest "sibling MessageSentEvent during sendHandler await does not corrupt state": ## Regression test for the prune-during-await race ## (PR #3914 review comment r3324891059). Locks in that a sibling - ## `MessageSentEvent` firing while `onReadyToSend` is paused at an - ## `await` does not lose the second `channelReqId`'s terminal - ## event. + ## `MessageSentEvent` firing while `send` is paused at a + ## `MessagingSend.request` await does not lose the second + ## `channelReqId`'s terminal event. const channelId = ChannelId("sm-race-channel") contentTopic = ContentTopic("/reliable-channel/test/sm-race") @@ -360,8 +360,8 @@ suite "Reliable Channel - send state machine": proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} = ## Call 2 fires the first segment's terminal event and then ## yields, so the listener task runs while the second segment - ## is still mid-`await` in `onReadyToSend` — the exact race - ## window the regression test targets. + ## is still mid-`await` inside `send` — the exact race window + ## the regression test targets. let id = RequestId("race-msg-req-" & $(msgReqIds.len + 1)) msgReqIds.add(id) if msgReqIds.len == 2: @@ -396,8 +396,7 @@ suite "Reliable Channel - send state machine": (await manager.send(channelId, "first".toBytes())).expect("send 1") ## Drain the first segment fully before queueing the second, so - ## the rate-limit FIFO between sibling sends isn't itself under - ## test here. + ## inter-send ordering isn't itself under test here. let firstDispatched = Moment.now() + 1.seconds while Moment.now() < firstDispatched and msgReqIds.len < 1: await sleepAsync(5.milliseconds) @@ -407,8 +406,8 @@ suite "Reliable Channel - send state machine": (await manager.send(channelId, "second".toBytes())).expect("send 2") ## Wait until `fakeSend(m2)` has fully returned and yield once - ## more so `onReadyToSend`'s post-await continuation gets a chance - ## to register `id2` in `inflightMessagingIds` before we emit its + ## more so `send`'s post-await continuation gets a chance to + ## register `id2` in `inflightMessagingIds` before we emit its ## terminal event. let dispatchDeadline = Moment.now() + 1.seconds while Moment.now() < dispatchDeadline and sendsReturned < 2: From f43399833882131913d4a8d8afde26dd646a766e Mon Sep 17 00:00:00 2001 From: stubbsta Date: Wed, 8 Jul 2026 13:59:37 +0200 Subject: [PATCH 2/3] 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 From c7f767d6c7da075bd17f8ba892fccd8e2920864f Mon Sep 17 00:00:00 2001 From: stubbsta Date: Wed, 8 Jul 2026 14:33:46 +0200 Subject: [PATCH 3/3] feat: meter SendService transmissions through RateLimitManager Wires the relocated RateLimitManager into the messaging layer at the transmission stage rather than the API entry point: - MessagingClientConf gains a rateLimit: RateLimitConfig field (defaulting to DefaultEpochPeriodSec / DefaultMessagesPerEpoch), and MessagingClient.new hands the constructed manager to SendService. - SendService consults admit() before the first transmission of a task, both in send() and in the retry loop. Re-publishes of an already-propagated message (firstPropagatedTime set) skip admission: they resend the same bytes, which reuse the same RLN proof/nullifier and consume no fresh epoch slot. - An over-budget task parks in the task cache as NextRoundRetry; the service loop re-admits it as the epoch budget frees up. The skeleton admit() is a pass-through, so behaviour is unchanged today. Gating transmissions instead of MessagingClient.send keeps SDS repair rebroadcasts free of API-entry rejection (SDS decides that a repair is needed; the transmission scheduler decides when it fits the budget) while every wire transmission still draws from one node-wide budget -- which network-side RLN enforcement applies to repairs regardless of any local bypass. It is also where RLN proof attachment must happen, since proofs bind to the epoch current at transmission time. Adds tests/messaging/test_rate_limit_manager.nim covering the current disabled + enabled pass-through behaviour, wired into all_tests_waku.nim. Co-Authored-By: Claude Opus 4.7 (1M context) Co-Authored-By: Claude Opus 4.8 (1M context) --- logos_delivery/api/conf/messaging_conf.nim | 6 ++++- .../send_service/send_service.nim | 25 ++++++++++++++++-- logos_delivery/messaging/messaging_client.nim | 6 +++-- tests/all_tests_waku.nim | 3 +++ tests/messaging/test_all.nim | 3 +++ tests/messaging/test_rate_limit_manager.nim | 26 +++++++++++++++++++ 6 files changed, 64 insertions(+), 5 deletions(-) create mode 100644 tests/messaging/test_all.nim create mode 100644 tests/messaging/test_rate_limit_manager.nim diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index a8c91acd7..0e2cf1786 100644 --- a/logos_delivery/api/conf/messaging_conf.nim +++ b/logos_delivery/api/conf/messaging_conf.nim @@ -4,8 +4,9 @@ import results, 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 -export kernel_conf +export kernel_conf, rate_limit_manager type LogosDeliveryMode* {.pure.} = enum Edge # client-only node @@ -52,6 +53,9 @@ type MessagingClientConf* = object ## Process log format (TEXT or JSON); applied by the kernel on node creation. nodeKey* {.name: "nodekey".}: Opt[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. 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/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim index 097bb0752..feec01953 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim @@ -8,7 +8,8 @@ import ./[send_processor, relay_processor, lightpush_processor, delivery_task], logos_delivery/waku/[waku_core, waku_store/common], logos_delivery/waku/waku, - logos_delivery/waku/api/[store, subscriptions, publish] + logos_delivery/waku/api/[store, subscriptions, publish], + logos_delivery/messaging/rate_limit_manager/rate_limit_manager import logos_delivery/api/events/messaging_client_events logScope: @@ -46,6 +47,9 @@ type SendService* = ref object of RootObj serviceLoopHandle: Future[void] ## handle that allows to stop the async task sendProcessor: BaseSendProcessor + rateLimitManager: RateLimitManager + ## Meters first transmissions against the per-epoch budget; re-publishes + ## of an already-propagated message resend the same bytes and are free. waku: Waku checkStoreForMessages: bool @@ -77,7 +81,10 @@ proc setupSendProcessorChain( return ok(processors[0]) proc new*( - T: typedesc[SendService], preferP2PReliability: bool, waku: Waku + T: typedesc[SendService], + preferP2PReliability: bool, + waku: Waku, + rateLimitManager: RateLimitManager, ): Result[T, string] = if not waku.hasRelay() and not waku.hasLightpush(): return err( @@ -94,6 +101,7 @@ proc new*( taskCache: newSeq[DeliveryTask](), serviceLoopHandle: nil, sendProcessor: sendProcessorChain, + rateLimitManager: rateLimitManager, waku: waku, checkStoreForMessages: checkStoreForMessages, lastStoreCheckTime: Moment.now(), @@ -245,6 +253,12 @@ proc trySendMessages(self: SendService) {.async.} = for task in tasksToSend: # Todo, check if it has any perf gain to run them concurrent... + if task.firstPropagatedTime.isNone(): + ## Never propagated: this transmission consumes a fresh epoch slot, so + ## it must be admitted. Tasks that stay over budget remain in + ## `NextRoundRetry` and are retried as the epoch rolls over. + (await self.rateLimitManager.admit(task.msg.payload)).isOkOr: + continue await self.sendProcessor.process(task) proc serviceLoop(self: SendService) {.async.} = @@ -274,6 +288,13 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} = error "SendService.send: failed to subscribe to content topic", contentTopic = task.msg.contentTopic, error = error + (await self.rateLimitManager.admit(task.msg.payload)).isOkOr: + info "SendService.send: over rate-limit budget, parking task", + requestId = task.requestId, msgHash = task.msgHash.to0xHex() + 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/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim index d883878fb..494ba5ad1 100644 --- a/logos_delivery/messaging/messaging_client.nim +++ b/logos_delivery/messaging/messaging_client.nim @@ -7,7 +7,8 @@ import logos_delivery/api/messaging_client_api, logos_delivery/waku/waku, logos_delivery/waku/factory/conf_builder/waku_conf_builder, - logos_delivery/messaging/delivery_service/[recv_service, send_service] + logos_delivery/messaging/delivery_service/[recv_service, send_service], + logos_delivery/messaging/rate_limit_manager/rate_limit_manager export messaging_client_api, messaging_conf @@ -24,7 +25,8 @@ proc new*( ## The messaging layer chains onto Waku: it drives the underlying Waku kernel ## for transport while exposing its own send/recv API. let reliability = conf.reliabilityEnabled.get(DefaultP2pReliability) - let sendService = ?SendService.new(reliability, waku) + let sendService = + ?SendService.new(reliability, waku, RateLimitManager.new(conf.rateLimit)) let recvService = RecvService.new(waku) return ok( T( diff --git a/tests/all_tests_waku.nim b/tests/all_tests_waku.nim index 5498db307..b3f72fe90 100644 --- a/tests/all_tests_waku.nim +++ b/tests/all_tests_waku.nim @@ -88,5 +88,8 @@ import ./tools/test_all # Persistency library tests import ./persistency/test_all +# Messaging API tests +import ./messaging/test_all + # Reliable Channel API tests import ./channels/test_all diff --git a/tests/messaging/test_all.nim b/tests/messaging/test_all.nim new file mode 100644 index 000000000..3cf508033 --- /dev/null +++ b/tests/messaging/test_all.nim @@ -0,0 +1,3 @@ +{.used.} + +import ./test_rate_limit_manager diff --git a/tests/messaging/test_rate_limit_manager.nim b/tests/messaging/test_rate_limit_manager.nim new file mode 100644 index 000000000..895fcb49d --- /dev/null +++ b/tests/messaging/test_rate_limit_manager.nim @@ -0,0 +1,26 @@ +{.used.} + +import chronos, testutils/unittests, stew/byteutils + +import logos_delivery/messaging/rate_limit_manager/rate_limit_manager + +suite "RateLimitManager - admission": + asyncTest "admit is a pass-through when disabled": + let rl = RateLimitManager.new( + RateLimitConfig(enabled: false, epochPeriodSec: 600, messagesPerEpoch: 1) + ) + for _ in 0 ..< 10: + 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. + 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()