mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-19 11:09:26 +00:00
Merge c7f767d6c7da075bd17f8ba892fccd8e2920864f into 9827be59905bb86abae59705202b5b16816bf805
This commit is contained in:
Tanya S
GitHub
commit
cc919d8e40
@ -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
|
||||
|
||||
# `LogosDeliveryMode` and `EntryLayer` are defined at the leaf (`cli_args`) so
|
||||
# they can appear on `WakuNodeConf`; re-exported here via `kernel_conf`.
|
||||
@ -50,6 +51,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.
|
||||
|
||||
@ -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,
|
||||
)
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
@ -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: []).} =
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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
|
||||
@ -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
|
||||
|
||||
@ -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:
|
||||
|
||||
3
tests/messaging/test_all.nim
Normal file
3
tests/messaging/test_all.nim
Normal file
@ -0,0 +1,3 @@
|
||||
{.used.}
|
||||
|
||||
import ./test_rate_limit_manager
|
||||
26
tests/messaging/test_rate_limit_manager.nim
Normal file
26
tests/messaging/test_rate_limit_manager.nim
Normal file
@ -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()
|
||||
Loading…
x
Reference in New Issue
Block a user