diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index a8c91acd76..0e2cf17867 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/channels/api/channel_lifecycle.nim b/logos_delivery/channels/api/channel_lifecycle.nim index 0d32e2acf2..57d4e83fe1 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 23c41ac13b..7af5fdca0e 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/rate_limit_manager/rate_limit_manager.nim b/logos_delivery/channels/rate_limit_manager/rate_limit_manager.nim deleted file mode 100644 index ab5a9f67bc..0000000000 --- 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/channels/reliable_channel.nim b/logos_delivery/channels/reliable_channel.nim index ca74f9df47..0286e8f27a 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/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim index 097bb07527..feec01953b 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 d883878fb1..494ba5ad1d 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/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 0000000000..7ee197d8a9 --- /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 diff --git a/tests/all_tests_waku.nim b/tests/all_tests_waku.nim index 5498db3073..b3f72fe903 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/channels/test_reliable_channel_send_receive.nim b/tests/channels/test_reliable_channel_send_receive.nim index 53a88f27a1..8a7b022310 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: diff --git a/tests/messaging/test_all.nim b/tests/messaging/test_all.nim new file mode 100644 index 0000000000..3cf508033f --- /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 0000000000..895fcb49dd --- /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()