start basic reliable channel folder (#3886)

Author: Ivansete-status

.github/workflows/ci.yml

@@ -43,6 +43,7 @@ jobs:
           - 'tools/**'
           - 'tests/all_tests_v2.nim'
           - 'tests/**'
+          - 'channels/**'
           docker:
           - 'docker/**'
 

channels/encryption/encryption.nim

@@ -0,0 +1,25 @@
+## Optional encryption hooks for the Reliable Channel API.
+##
+## Modelled as `RequestBroker`s: the broker pattern lets the channel
+## delegate work to a provider that may live in any module without
+## introducing a direct dependency. If no provider is registered the
+## broker returns an error, so installing the noop providers from
+## `noop_encryption` is required when the application does not want
+## actual encryption.
+##
+## Applied per-segment after SDS processing on outgoing, and before
+## SDS processing on incoming. No specific scheme is mandated.
+##
+## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html
+
+import brokers/request_broker
+
+export request_broker
+
+RequestBroker:
+  type Encrypt* = seq[byte]
+  proc signature*(payload: seq[byte]): Future[Result[Encrypt, string]] {.async.}
+
+RequestBroker:
+  type Decrypt* = seq[byte]
+  proc signature*(payload: seq[byte]): Future[Result[Decrypt, string]] {.async.}

channels/encryption/noop_encryption.nim

@@ -0,0 +1,18 @@
+## No-op encryption providers. Install these when the application does
+## not want actual encryption so the `Encrypt` / `Decrypt` brokers have
+## something to dispatch to.
+
+import results
+import chronos
+import ./encryption
+
+proc setNoopEncryption*() =
+  discard Encrypt.setProvider(
+    proc(payload: seq[byte]): Future[Result[Encrypt, string]] {.async.} =
+      return ok(Encrypt(payload))
+  )
+
+  discard Decrypt.setProvider(
+    proc(payload: seq[byte]): Future[Result[Decrypt, string]] {.async.} =
+      return ok(Decrypt(payload))
+  )

channels/events.nim

@@ -0,0 +1,23 @@
+## Reliable Channel event types emitted to API consumers.
+##
+## Lifecycle events for individual segments (sent / propagated / errored)
+## are the same as the network-level ones the DeliveryService already
+## emits — `requestId` is shared across layers — so we just re-export
+## `waku/events/message_events` and avoid declaring duplicates.
+##
+## Only the channel-level `MessageReceivedEvent` carries data that has
+## no analogue in the lower layer (reassembled application payload,
+## senderId, channelId), so it lives here.
+
+import waku/events/message_events as waku_message_events
+import brokers/event_broker
+
+import ./types as channel_types
+
+export waku_message_events, channel_types, event_broker
+
+EventBroker:
+  type ChannelMessageReceivedEvent* = object
+    channelId*: ChannelId
+    senderId*: SdsParticipantID
+    payload*: seq[byte]

channels/rate_limit_manager/rate_limit_manager.nim

@@ -0,0 +1,80 @@
+## 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* = 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