From c7f767d6c7da075bd17f8ba892fccd8e2920864f Mon Sep 17 00:00:00 2001 From: stubbsta Date: Wed, 8 Jul 2026 14:33:46 +0200 Subject: [PATCH] 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()