From 1ffee9600c60e97b05564a441ad78982dae8f312 Mon Sep 17 00:00:00 2001 From: Ivan FB Date: Wed, 1 Jul 2026 20:28:53 +0200 Subject: [PATCH] avoid sendHandler with concepts --- logos_delivery/api/messaging_client_api.nim | 11 +- .../api/reliable_channel_manager_api.nim | 18 +- .../channels/api/channel_lifecycle.nim | 17 +- logos_delivery/channels/reliable_channel.nim | 46 ++--- .../channels/reliable_channel_manager.nim | 40 ++--- logos_delivery/logos_delivery.nim | 4 +- .../test_reliable_channel_send_receive.nim | 167 ++++++++++-------- 7 files changed, 152 insertions(+), 151 deletions(-) diff --git a/logos_delivery/api/messaging_client_api.nim b/logos_delivery/api/messaging_client_api.nim index 536ce94fa..fe8e6a286 100644 --- a/logos_delivery/api/messaging_client_api.nim +++ b/logos_delivery/api/messaging_client_api.nim @@ -4,8 +4,17 @@ import logos_delivery/api/types as api_types export api_types +type MessagingSender* = concept c + ## The narrow egress capability the reliable-channel layer depends on: just + ## `send`. A layer above messaging depends on this capability, not on the + ## concrete `MessagingClient`. Dot-call form (`c.send(...)`) so it is + ## satisfied by both a real `send` proc (the production `MessagingClient`) + ## and a `send` proc-typed field (a test double). + c.send(MessageEnvelope) is Future[Result[RequestId, string]] + # Structural API contract for a messaging client (ops in `messaging/api/*`). +# Refines `MessagingSender`, so the send capability is stated once. type MessagingApi* = concept c + c is MessagingSender subscribe(c, contentTopic = ContentTopic) is Future[Result[void, string]] unsubscribe(c, contentTopic = ContentTopic) is Result[void, string] - send(c, envelope = MessageEnvelope) is Future[Result[RequestId, string]] diff --git a/logos_delivery/api/reliable_channel_manager_api.nim b/logos_delivery/api/reliable_channel_manager_api.nim index d50b57288..08d0f268a 100644 --- a/logos_delivery/api/reliable_channel_manager_api.nim +++ b/logos_delivery/api/reliable_channel_manager_api.nim @@ -1,19 +1,17 @@ import chronos, results import logos_delivery/api/types as api_types +import logos_delivery/api/messaging_client_api import logos_delivery/channels/types as channel_types -export api_types, channel_types +# `messaging_client_api` is re-exported for `MessagingSender`, the egress +# capability the generic `ReliableChannel[M]`/`ReliableChannelManager[M]` need. +export api_types, messaging_client_api, channel_types -type SendHandler* = proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {. - async: (raises: [CatchableError]), gcsafe -.} - ## Egress dispatch boundary. Typically wraps `MessagingClient.send`; - ## tests inject a fake that records calls and returns canned - ## `RequestId`s so the send state machine can be exercised end-to-end - ## without a network. - -# Structural API contract for the reliable-channel surface (ops in `channels/api/*`). +# Structural API contract for the reliable-channel surface. This is the +# node-bound consumer view (the messaging node is already bound), so it is +# satisfied by `LogosDelivery` — the concentrator that owns the node — rather +# than by the manager, whose `createReliableChannel` takes the node explicitly. type ReliableChannelApi* = concept c createReliableChannel( c, channelId = ChannelId, contentTopic = ContentTopic, senderId = SdsParticipantID diff --git a/logos_delivery/channels/api/channel_lifecycle.nim b/logos_delivery/channels/api/channel_lifecycle.nim index b50ba1b63..5e17641ec 100644 --- a/logos_delivery/channels/api/channel_lifecycle.nim +++ b/logos_delivery/channels/api/channel_lifecycle.nim @@ -8,7 +8,7 @@ import logos_delivery/channels/reliable_channel_manager import logos_delivery/channels/reliable_channel import logos_delivery/waku/persistency/sds_persistency -# ReliableChannel, SendHandler, config and wire-version markers. +# ReliableChannel, config and wire-version markers. export reliable_channel const SdsJobId = "sds" @@ -32,16 +32,9 @@ proc createReliableChannel*( channelId: ChannelId, contentTopic: ContentTopic, senderId: SdsParticipantID, - sendHandler: SendHandler = nil, ): Result[ChannelId, string] = - ## Spec entry point. The `sendHandler` and `rng` the channel needs are - ## sourced from the owning `ReliableChannelManager` rather than passed - ## per call. Encryption is wired up through the `Encrypt`/`Decrypt` - ## request brokers — the application installs its own providers - ## (or `setNoopEncryption()`) before traffic flows. - ## - ## `sendHandler` defaults to the manager's default (constructed at mount - ## from `MessagingClient.send`); tests pass a fake to bypass the network. + ## Spec entry point. The messaging node the channel dispatches through is the + ## one the manager owns (`self.messaging`), so callers address channels by id. if self.channels.hasKey(channelId): return err("channel already exists: " & channelId) @@ -60,10 +53,8 @@ proc createReliableChannel*( epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch ) - let effectiveSendHandler = if sendHandler.isNil(): self.sendHandler else: sendHandler - let chn = ReliableChannel.new( - sendHandler = effectiveSendHandler, + messaging = self.messaging, channelId = channelId, contentTopic = contentTopic, senderId = senderId, diff --git a/logos_delivery/channels/reliable_channel.nim b/logos_delivery/channels/reliable_channel.nim index 06648a8f7..904cdc060 100644 --- a/logos_delivery/channels/reliable_channel.nim +++ b/logos_delivery/channels/reliable_channel.nim @@ -21,11 +21,11 @@ import bearssl/rand import stew/byteutils import libp2p/crypto/crypto as libp2p_crypto +import brokers/broker_context import logos_delivery/api/types import logos_delivery/api/reliable_channel_manager_api import logos_delivery/api/events/messaging_client_events import logos_delivery/api/events/reliable_channel_manager_events -import logos_delivery/messaging/delivery_service/send_service import logos_delivery/waku/waku_core/topics import ./segmentation/segmentation @@ -34,7 +34,7 @@ import ./rate_limit_manager/rate_limit_manager import ./encryption/encryption export - types, reliable_channel_manager_api, send_service, segmentation, scalable_data_sync, + types, reliable_channel_manager_api, segmentation, scalable_data_sync, rate_limit_manager, encryption const LipWireReliableChannelVersion* = "RELIABLE-CHANNEL-API/1" @@ -80,11 +80,15 @@ type ## and that scan is correct precisely because the outer iteration ## order matches the order `send` pushed entries. - ReliableChannel* = ref object + ReliableChannel*[M: MessagingSender] = ref object ## Spec-defined public type. Fields are private so callers cannot ## mutate internals and break invariants. Getters are added below ## for the few values consumers may need. - sendHandler: SendHandler + ## + ## Generic over `M`, the messaging egress: the channel calls `M.send` + ## directly (compile-time dispatch). Production + ## instantiates `M = MessagingClient`; tests instantiate a one-proc double. + messaging: M channelId: ChannelId contentTopic: ContentTopic senderId: SdsParticipantID @@ -195,7 +199,7 @@ proc onMessageFinal( proc markSegmentFailed(self: ReliableChannel, channelReqId: RequestId) = try: self.channelReqs[channelReqId].failedCount.inc() - except KeyError as e: + except system.KeyError as e: error "unreachable: channelReqId not found in markSegmentFailed", channelReqId = $channelReqId, error = e.msg return @@ -206,7 +210,7 @@ proc markSegmentInflight( ) = try: self.channelReqs[channelReqId].inflightMessagingIds.add(messagingReqId) - except KeyError as e: + except system.KeyError as e: error "unreachable: channelReqId not found in markSegmentInflight", channelReqId = $channelReqId, error = e.msg @@ -263,12 +267,12 @@ proc onReadyToSend( meta: LipWireReliableChannelVersion.toBytes(), ) - ## `sendHandler` is not annotated `(raises: [])`, but this listener is. - ## Convert any raise to a Result error so the state machine handles - ## both failure modes (Result.err and exception) through one path. + ## `M.send` (e.g. `MessagingClient.send`) is not annotated `(raises: [])`, + ## but this listener is. Convert any raise to a Result error so the state + ## machine handles both failure modes through one path. let sendRes = try: - await self.sendHandler(envelope) + await self.messaging.send(envelope) except CatchableError as e: Result[RequestId, string].err("messaging send raised: " & e.msg) @@ -360,14 +364,16 @@ proc dispatchRepair(self: ReliableChannel, wire: seq[byte]) {.async: (raises: [] meta: LipWireReliableChannelVersion.toBytes(), ) + ## `M.send` may raise; this proc is `(raises: [])`, so funnel any raise into + ## a Result the same way the main dispatch path does. let sendRes = try: - await self.sendHandler(envelope) + await self.messaging.send(envelope) except CatchableError as e: Result[RequestId, string].err("messaging send raised: " & e.msg) - if sendRes.isErr(): + sendRes.isOkOr: debug "SDS repair rebroadcast dropped: dispatch failed", - channelId = self.channelId, error = sendRes.error + channelId = self.channelId, error = error proc onMessageReceived( self: ReliableChannel, messageHash: string, payload: seq[byte] @@ -413,9 +419,9 @@ proc onMessageReceived( for content in deliverable: self.reportReceived(content) -proc new*( +proc new*[M: MessagingSender]( T: type ReliableChannel, - sendHandler: SendHandler, + messaging: M, channelId: ChannelId, contentTopic: ContentTopic, senderId: SdsParticipantID, @@ -423,19 +429,15 @@ proc new*( sdsConfig: SdsConfig, rateConfig: RateLimitConfig, brokerCtx: BrokerContext = globalBrokerContext(), -): T = +): ReliableChannel[M] = ## 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. - ## - ## `sendHandler` is the egress dispatch. The owning `ReliableChannelManager` - ## typically constructs it as a closure over `MessagingClient.send`. Tests - ## pass a fake to drive the send state machine without touching the network. - let chn = T( - sendHandler: sendHandler, + let chn = ReliableChannel[M]( + messaging: messaging, channelId: channelId, contentTopic: contentTopic, senderId: senderId, diff --git a/logos_delivery/channels/reliable_channel_manager.nim b/logos_delivery/channels/reliable_channel_manager.nim index e8d579326..a0c260234 100644 --- a/logos_delivery/channels/reliable_channel_manager.nim +++ b/logos_delivery/channels/reliable_channel_manager.nim @@ -13,8 +13,6 @@ import stew/byteutils import brokers/broker_context -import logos_delivery/messaging/messaging_client -import logos_delivery/messaging/api/send import logos_delivery/api/types import logos_delivery/api/reliable_channel_manager_api @@ -28,37 +26,25 @@ type ## channel API. Placeholder for now (segmentation / SDS / rate-limit defaults ## will move here in a follow-up PR); kept so each layer owns its own config. - ReliableChannelManager* = ref object ## Implements `ReliableChannelApi`. - channels*: Table[ChannelId, ReliableChannel] ## read by `channels/api.nim` - messagingClient: MessagingClient ## The channel layer chains onto messaging. - sendHandler*: SendHandler - ## Default egress dispatch for channels created through this manager. - ## Built in `new` as a closure over `MessagingClient.send` so the channel - ## layer itself stays callable-only. + ReliableChannelManager*[M: MessagingSender] = ref object + ## Owns the set of channels and the messaging node they dispatch through. + ## Generic over the egress `M`; hands the node to every channel it creates. + ## Holding the node here keeps `createReliableChannel` node-free — the + ## node-bound consumer surface the `ReliableChannelApi` concept describes. + channels*: Table[ChannelId, ReliableChannel[M]] ## read by `channels/api.nim` + messaging*: M ## egress node, passed on to each `ReliableChannel` brokerCtx*: BrokerContext -proc new*( +proc new*[M: MessagingSender]( T: type ReliableChannelManager, conf: ReliableChannelManagerConf, - messagingClient: MessagingClient, + messaging: M, brokerCtx: BrokerContext = globalBrokerContext(), -): Result[T, string] = - ## The reliable channel layer chains onto the messaging layer: its default - ## egress is `MessagingClient.send`, wrapped here so callers never wire the - ## handler themselves. - if messagingClient.isNil(): - return err("messaging client is required") - - let defaultSendHandler: SendHandler = proc( - envelope: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = - return await messagingClient.send(envelope) - +): Result[ReliableChannelManager[M], string] = return ok( - T( - channels: initTable[ChannelId, ReliableChannel](), - messagingClient: messagingClient, - sendHandler: defaultSendHandler, + ReliableChannelManager[M]( + channels: initTable[ChannelId, ReliableChannel[M]](), + messaging: messaging, brokerCtx: brokerCtx, ) ) diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim index 0b161d6f7..eebeef551 100644 --- a/logos_delivery/logos_delivery.nim +++ b/logos_delivery/logos_delivery.nim @@ -74,7 +74,7 @@ type LogosDelivery* = ref object ## Entry point. Holds one instance of each API layer. waku*: Waku messagingClient*: MessagingClient - reliableChannelManager*: ReliableChannelManager + reliableChannelManager*: ReliableChannelManager[MessagingClient] proc init*(T: type LogosDeliveryConf, wakuConf: WakuConf): LogosDeliveryConf = ## Builds the aggregated config from a `WakuConf`. The messaging / reliable @@ -154,5 +154,5 @@ proc isOnline*(self: LogosDelivery): Future[Result[bool, string]] {.async.} = static: doAssert Waku is KernelApi doAssert MessagingClient is MessagingApi - doAssert ReliableChannelManager is ReliableChannelApi + doAssert ReliableChannelManager[MessagingClient] is ReliableChannelApi doAssert LogosDelivery is LogosDeliveryApi diff --git a/tests/channels/test_reliable_channel_send_receive.nim b/tests/channels/test_reliable_channel_send_receive.nim index c20f0d0b0..0a74f2d87 100644 --- a/tests/channels/test_reliable_channel_send_receive.nim +++ b/tests/channels/test_reliable_channel_send_receive.nim @@ -38,6 +38,27 @@ proc createApiNodeConf(): WakuNodeConf = conf.rest = false return conf +type MessagingClientTest = ref object + ## Test double for the messaging node used by the send-side tests. `send` is + ## a proc-typed field, so it satisfies the (dot-form) `MessagingSender` + ## concept directly; the manager owns it as `manager.messaging`, and each + ## test scripts the egress with `manager.messaging.send = ...`. + send: proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {. + async: (raises: [CatchableError]), gcsafe + .} + +static: + doAssert MessagingClientTest is MessagingSender + +proc newTestManager( + brokerCtx: BrokerContext +): ReliableChannelManager[MessagingClientTest] = + ## Manager over the send double. `manager.messaging` is the double, so tests + ## script the egress with `manager.messaging.send = ...`. + ReliableChannelManager + .new(ReliableChannelManagerConf(), MessagingClientTest(), brokerCtx) + .expect("ReliableChannelManager.new") + suite "Reliable Channel - ingress": asyncTest "manager dispatches marked WakuMessage to the right channel": ## Unit test for the receive side of the API: instead of standing @@ -54,7 +75,7 @@ suite "Reliable Channel - ingress": let appPayload = "hello reliable channel".toBytes() var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -120,7 +141,7 @@ suite "Reliable Channel - ingress": let appPayload = "foreign payload".toBytes() var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -165,8 +186,8 @@ suite "Reliable Channel - ingress": suite "Reliable Channel - send state machine": asyncTest "MessageSentEvent finalises the channelReqId as Sent": ## Drives the real send pipeline (`send` -> segmentation -> SDS -> - ## rate_limit -> encrypt -> dispatch) via a fake `SendHandler` that - ## returns a canned `RequestId` instead of hitting the network. + ## rate_limit -> encrypt -> dispatch) via the `MessagingClientTest` double + ## that returns a canned `RequestId` instead of hitting the network. ## Emitting the delivery-layer `MessageSentEvent` must drive the ## channel-level state machine through `Confirmed` and produce a ## `ChannelMessageSentEvent` (channel-level terminal event) for the @@ -177,26 +198,24 @@ suite "Reliable Channel - send state machine": fakeMsgReqId = RequestId("fake-msg-req-1") var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClientTest] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") - manager = waku.reliableChannelManager + manager = newTestManager(brokerCtx) setNoopEncryption() var sendCalls = 0 - let fakeSend: SendHandler = proc( - env: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = + manager.messaging.send = proc( + envelope: MessageEnvelope + ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} = sendCalls.inc return ok(fakeMsgReqId) discard manager - .createReliableChannel( - channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend - ) + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) .expect("createReliableChannel") let sentFut = newFuture[RequestId]("channel-sent") @@ -227,13 +246,14 @@ suite "Reliable Channel - send state machine": if finalised: check sentFut.read() == channelReqId + await manager.stop() (await waku.stop()).expect("stop") asyncTest "two independent channelReqIds are finalised independently": ## Two `send()` calls -> two independent `channelReqId`s, each with ## one segment under the current segmentation skeleton ## (`performSegmentation` always emits exactly one segment). The - ## fake `SendHandler` returns distinct `messagingReqId`s; finalising + ## send double returns distinct `messagingReqId`s; finalising ## the first emits `ChannelMessageSentEvent` for its `channelReqId`, ## finalising the second as a failure emits `ChannelMessageErrorEvent` ## for the other. @@ -242,27 +262,25 @@ suite "Reliable Channel - send state machine": contentTopic = ContentTopic("/reliable-channel/test/sm-multi") var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClientTest] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") - manager = waku.reliableChannelManager + manager = newTestManager(brokerCtx) setNoopEncryption() var msgReqIds: seq[RequestId] - let fakeSend: SendHandler = proc( - env: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = + manager.messaging.send = proc( + envelope: MessageEnvelope + ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} = let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1)) msgReqIds.add(id) return ok(id) discard manager - .createReliableChannel( - channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend - ) + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) .expect("createReliableChannel") let sentFut = newFuture[RequestId]("channel-sent") @@ -319,6 +337,7 @@ suite "Reliable Channel - send state machine": if erroredArrived: check erroredFut.read() == channelReqId2 + await manager.stop() (await waku.stop()).expect("stop") asyncTest "TODO: channelReqId not pruned until ALL its segments are final": @@ -328,11 +347,11 @@ suite "Reliable Channel - send state machine": ## `channelReqId` cannot be produced through the real pipeline. ## Implement once segmentation does real chunking: send a payload ## larger than `DefaultSegmentSizeBytes`, capture the N - ## `messagingReqId`s from a fake `SendHandler`, finalise some, and + ## `messagingReqId`s from the send double, finalise some, and ## assert prune only fires once every sibling is final. skip() - asyncTest "sibling MessageSentEvent during sendHandler await does not corrupt state": + asyncTest "sibling MessageSentEvent during send 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 @@ -343,20 +362,20 @@ suite "Reliable Channel - send state machine": contentTopic = ContentTopic("/reliable-channel/test/sm-race") var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClientTest] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") - manager = waku.reliableChannelManager + manager = newTestManager(brokerCtx) setNoopEncryption() var msgReqIds: seq[RequestId] var sendsReturned = 0 - let fakeSend: SendHandler = proc( - env: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = + manager.messaging.send = proc( + envelope: MessageEnvelope + ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} = ## 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 @@ -364,18 +383,15 @@ suite "Reliable Channel - send state machine": let id = RequestId("race-msg-req-" & $(msgReqIds.len + 1)) msgReqIds.add(id) if msgReqIds.len == 2: - waku_message_events.MessageSentEvent.emit( - brokerCtx, - waku_message_events.MessageSentEvent(requestId: msgReqIds[0], messageHash: ""), + MessageSentEvent.emit( + brokerCtx, MessageSentEvent(requestId: msgReqIds[0], messageHash: "") ) await sleepAsync(50.milliseconds) sendsReturned.inc() return ok(id) discard manager - .createReliableChannel( - channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend - ) + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) .expect("createReliableChannel") var finalisedReqIds: seq[RequestId] @@ -431,6 +447,7 @@ suite "Reliable Channel - send state machine": check channelReqId1 in finalisedReqIds check channelReqId2 in finalisedReqIds + await manager.stop() (await waku.stop()).expect("stop") suite "Reliable Channel - SDS persistence": @@ -450,22 +467,20 @@ suite "Reliable Channel - SDS persistence": removeDir(root) var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClientTest] lockNewGlobalBrokerContext: waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") - manager = waku.reliableChannelManager + manager = newTestManager(globalBrokerContext()) setNoopEncryption() - let fakeSend: SendHandler = proc( - env: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = + manager.messaging.send = proc( + envelope: MessageEnvelope + ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} = return ok(RequestId("persist-msg-req-1")) discard manager - .createReliableChannel( - channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend - ) + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) .expect("createReliableChannel") discard (await manager.send(channelId, "persist me".toBytes())).expect("send") @@ -496,6 +511,7 @@ suite "Reliable Channel - SDS persistence": check await pollMetaExists() check await pollLogRow() + await manager.stop() (await waku.stop()).expect("stop") ## A marked WakuMessage carrying an SDS envelope, as it arrives off the wire. @@ -518,7 +534,7 @@ suite "Reliable Channel - SDS lifecycle": let payload2 = "second message".toBytes() var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -589,7 +605,7 @@ suite "Reliable Channel - SDS lifecycle": let appPayload = "deliver once".toBytes() var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -646,7 +662,7 @@ suite "Reliable Channel - SDS lifecycle": contentTopic = ContentTopic("/reliable-channel/test/foreign") var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -706,7 +722,7 @@ suite "Reliable Channel - SDS lifecycle": removeDir(root) var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -788,27 +804,25 @@ suite "Reliable Channel - SDS protocol semantics": contentTopic = ContentTopic("/reliable-channel/test/semantics") var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClientTest] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") - manager = waku.reliableChannelManager + manager = newTestManager(brokerCtx) setNoopEncryption() var capturedWires: seq[seq[byte]] - let fakeSend: SendHandler = proc( - env: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = + manager.messaging.send = proc( + envelope: MessageEnvelope + ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} = ## Noop encryption is identity, so the envelope payload IS the SDS wire. - capturedWires.add(env.payload) + capturedWires.add(envelope.payload) return ok(RequestId("semantics-req-" & $capturedWires.len)) discard manager - .createReliableChannel( - channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend - ) + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) .expect("createReliableChannel") let remotePeer = @@ -838,6 +852,7 @@ suite "Reliable Channel - SDS protocol semantics": check SdsMessageID("semantics-m1") in reply.causalHistory.getMessageIds() check reply.lamportTimestamp > m1.lamportTimestamp + await manager.stop() (await waku.stop()).expect("stop") asyncTest "an unacknowledged send is acked by a later remote message": @@ -856,26 +871,25 @@ suite "Reliable Channel - SDS protocol semantics": removeDir(root) var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClientTest] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") - manager = waku.reliableChannelManager + manager = newTestManager(brokerCtx) setNoopEncryption() var capturedWires: seq[seq[byte]] - let fakeSend: SendHandler = proc( - env: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = - capturedWires.add(env.payload) + manager.messaging.send = proc( + envelope: MessageEnvelope + ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} = + ## Noop encryption is identity, so the envelope payload IS the SDS wire. + capturedWires.add(envelope.payload) return ok(RequestId("ack-req-" & $capturedWires.len)) discard manager - .createReliableChannel( - channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend - ) + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) .expect("createReliableChannel") discard (await manager.send(channelId, "needs ack".toBytes())).expect("send") @@ -932,6 +946,7 @@ suite "Reliable Channel - SDS protocol semantics": await sleepAsync(5.milliseconds) check bufLen == 0 + await manager.stop() (await waku.stop()).expect("stop") asyncTest "three-deep dependency chain is released in causal order": @@ -944,7 +959,7 @@ suite "Reliable Channel - SDS protocol semantics": @["chain first".toBytes(), "chain second".toBytes(), "chain third".toBytes()] var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -1019,7 +1034,7 @@ suite "Reliable Channel - SDS protocol semantics": let appPayload = "real message".toBytes() var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() @@ -1092,26 +1107,25 @@ suite "Reliable Channel - SDS protocol semantics": let appPayload = "ok".toBytes() var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClientTest] var brokerCtx: BrokerContext lockNewGlobalBrokerContext: brokerCtx = globalBrokerContext() waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") - manager = waku.reliableChannelManager + manager = newTestManager(brokerCtx) setNoopEncryption() var capturedWires: seq[seq[byte]] - let fakeSend: SendHandler = proc( - env: MessageEnvelope - ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} = - capturedWires.add(env.payload) + manager.messaging.send = proc( + envelope: MessageEnvelope + ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]).} = + ## Noop encryption is identity, so the envelope payload IS the SDS wire. + capturedWires.add(envelope.payload) return ok(RequestId("unique-req-" & $capturedWires.len)) discard manager - .createReliableChannel( - channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend - ) + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) .expect("createReliableChannel") var deliveries: seq[seq[byte]] @@ -1156,11 +1170,12 @@ suite "Reliable Channel - SDS protocol semantics": await sleepAsync(5.milliseconds) check deliveries.len == 2 + await manager.stop() (await waku.stop()).expect("stop") asyncTest "manager rejects operations on unknown channels": var waku: LogosDelivery - var manager: ReliableChannelManager + var manager: ReliableChannelManager[MessagingClient] lockNewGlobalBrokerContext: waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new") manager = waku.reliableChannelManager