chore: replace ReliableChannel sendhandler hack (#3994)

* squash original commits and rebase to master: removed sendHandler and replaced by MessagingSend BrokerRequest

* Needed to adapt to new messaging source structure to avoid cyclic imports

* Relocate MessagingSend request broker and eliminate api/requests

* fix ci fails

logos_delivery.nimble

@@ -65,7 +65,7 @@ requires "https://github.com/logos-messaging/nim-ffi#v0.1.3"
 
 requires "https://github.com/logos-messaging/nim-sds.git#b12f5ee07c5b764303b51fb948b32a4ade1de3b5"
 
-requires "https://github.com/NagyZoltanPeter/nim-brokers.git#v3.1.1"
+requires "https://github.com/NagyZoltanPeter/nim-brokers.git#v3.1.4"
 
 requires "https://github.com/vacp2p/nim-lsquic.git#v0.5.1"
 requires "https://github.com/vacp2p/nim-jwt.git#057ec95eb5af0eea9c49bfe9025b3312c95dc5f2"

logos_delivery/api/messaging_client_api.nim

@@ -1,11 +1,18 @@
 import chronos, results
-
+import brokers/[request_broker, broker_context]
 import logos_delivery/api/types as api_types
 
-export api_types
+export api_types, request_broker, broker_context
 
 # Structural API contract for a messaging client (ops in `messaging/api/*`).
 type MessagingApi* = concept c
   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]]
+
+# Semi detached MessagingClient send interface.
+# Can be used without MessagingClient, under the same context of MessagingClient instance.
+RequestBroker:
+  proc MessagingSend(
+    envelope: MessageEnvelope
+  ): Future[Result[RequestId, string]] {.async.}

logos_delivery/api/reliable_channel_manager_api.nim

@@ -5,14 +5,6 @@ import logos_delivery/channels/types as channel_types
 
 export api_types, 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/*`).
 type ReliableChannelApi* = concept c
   createReliableChannel(

logos_delivery/channels/api/channel_lifecycle.nim

@@ -32,16 +32,7 @@ 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.
   if self.channels.hasKey(channelId):
     return err("channel already exists: " & channelId)
 
@@ -60,10 +51,7 @@ proc createReliableChannel*(
     epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch
   )
 
-  let effectiveSendHandler = if sendHandler.isNil(): self.sendHandler else: sendHandler
-
   let chn = ReliableChannel.new(
-    sendHandler = effectiveSendHandler,
     channelId = channelId,
     contentTopic = contentTopic,
     senderId = senderId,

logos_delivery/channels/reliable_channel.nim

@@ -24,8 +24,9 @@ import libp2p/crypto/crypto as libp2p_crypto
 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/messaging_client_api
 import logos_delivery/api/events/reliable_channel_manager_events
-import logos_delivery/messaging/delivery_service/send_service
+import logos_delivery/messaging/messaging_client
 import logos_delivery/waku/waku_core/topics
 
 import ./segmentation/segmentation
@@ -34,7 +35,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"
@@ -84,7 +85,6 @@ type
     ## 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
     channelId: ChannelId
     contentTopic: ContentTopic
     senderId: SdsParticipantID
@@ -263,14 +263,7 @@ 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.
-    let sendRes =
-      try:
-        await self.sendHandler(envelope)
-      except CatchableError as e:
-        Result[RequestId, string].err("messaging send raised: " & e.msg)
+    let sendRes = await MessagingSend.request(self.brokerCtx, envelope)
 
     let messagingReqId = sendRes.valueOr:
       MessageErrorEvent.emit(
@@ -360,14 +353,9 @@ proc dispatchRepair(self: ReliableChannel, wire: seq[byte]) {.async: (raises: []
     meta: LipWireReliableChannelVersion.toBytes(),
   )
 
-  let sendRes =
-    try:
-      await self.sendHandler(envelope)
-    except CatchableError as e:
-      Result[RequestId, string].err("messaging send raised: " & e.msg)
-  if sendRes.isErr():
+  (await MessagingSend.request(self.brokerCtx, envelope)).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]
@@ -415,7 +403,6 @@ proc onMessageReceived(
 
 proc new*(
     T: type ReliableChannel,
-    sendHandler: SendHandler,
     channelId: ChannelId,
     contentTopic: ContentTopic,
     senderId: SdsParticipantID,
@@ -430,12 +417,7 @@ proc new*(
   ## 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,
     channelId: channelId,
     contentTopic: contentTopic,
     senderId: senderId,

logos_delivery/channels/reliable_channel_manager.nim

@@ -30,38 +30,14 @@ type
 
   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.
     brokerCtx*: BrokerContext
 
 proc new*(
     T: type ReliableChannelManager,
     conf: ReliableChannelManagerConf,
-    messagingClient: MessagingClient,
     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)
-
-  return ok(
-    T(
-      channels: initTable[ChannelId, ReliableChannel](),
-      messagingClient: messagingClient,
-      sendHandler: defaultSendHandler,
-      brokerCtx: brokerCtx,
-    )
-  )
+  return ok(T(channels: initTable[ChannelId, ReliableChannel](), brokerCtx: brokerCtx))
 
 proc start*(self: ReliableChannelManager): Result[void, string] =
   ## Placeholder: per-channel listeners are installed in `ReliableChannel.new`,

logos_delivery/logos_delivery.nim

@@ -35,7 +35,7 @@ import logos_delivery/waku/api/events/health_events
 export health_events
 
 # Messaging layer
-import logos_delivery/messaging/messaging_client
+import logos_delivery/messaging/[messaging_client, messaging_client_lifecycle]
 export messaging_client
 import logos_delivery/messaging/api/[subscription, send]
 export subscription, send
@@ -103,7 +103,7 @@ proc new*(
     return err("failed to create MessagingClient: " & error)
 
   let reliableChannelManager = ReliableChannelManager.new(
-    layerConf.reliableChannel, messagingClient, waku.brokerCtx
+    layerConf.reliableChannel, waku.brokerCtx
   ).valueOr:
     return err("failed to create ReliableChannelManager: " & error)
 

logos_delivery/messaging/messaging_client.nim

@@ -2,6 +2,7 @@
 ## lifecycle. The public operations (subscribe / unsubscribe / send) live in
 ## `messaging/api.nim`.
 import results, chronos
+import chronicles
 import
   logos_delivery/api/messaging_client_api,
   logos_delivery/waku/waku,
@@ -17,11 +18,12 @@ type
     ## follow-up PR. Today it only carries the p2p reliability toggle.
     useP2PReliability*: bool
 
-  MessagingClient* = ref object ## Implements `MessagingApi`.
+  MessagingClient* = ref object
+    brokerCtx*: BrokerContext
     waku*: Waku ## The Waku kernel this layer drives; read by `messaging/api/*`.
     sendService*: SendService
     recvService*: RecvService
-    started: bool
+    started*: bool
 
 proc new*(
     T: type MessagingClient, conf: MessagingClientConf, waku: Waku
@@ -30,22 +32,14 @@ proc new*(
   ## for transport while exposing its own send/recv API.
   let sendService = ?SendService.new(conf.useP2PReliability, waku)
   let recvService = RecvService.new(waku)
-  return ok(T(waku: waku, sendService: sendService, recvService: recvService))
-
-proc start*(self: MessagingClient): Result[void, string] =
-  if self.started:
-    return ok()
-  self.recvService.startRecvService()
-  self.sendService.startSendService()
-  self.started = true
-  ok()
-
-proc stop*(self: MessagingClient) {.async.} =
-  if not self.started:
-    return
-  await self.sendService.stopSendService()
-  await self.recvService.stopRecvService()
-  self.started = false
+  return ok(
+    T(
+      waku: waku,
+      sendService: sendService,
+      recvService: recvService,
+      brokerCtx: waku.brokerCtx,
+    )
+  )
 
 proc checkApiAvailability*(self: MessagingClient): Result[void, string] =
   ## Shared guard for the api operation module.

logos_delivery/messaging/messaging_client_lifecycle.nim

@@ -0,0 +1,37 @@
+## Messaging layer core: the `MessagingClient` type plus its construction and
+## lifecycle. The public operations (subscribe / unsubscribe / send) live in
+## `messaging/api.nim`.
+import results, chronos
+import chronicles
+import
+  logos_delivery/messaging/messaging_client,
+  logos_delivery/messaging/api/send,
+  logos_delivery/api/messaging_client_api,
+  logos_delivery/waku/waku,
+  logos_delivery/messaging/delivery_service/[recv_service, send_service]
+
+# Surfaces the messaging API interface (and its Message* events) to consumers.
+export messaging_client
+
+proc start*(self: MessagingClient): Result[void, string] =
+  if self.started:
+    return ok()
+  self.recvService.startRecvService()
+  self.sendService.startSendService()
+
+  ?MessagingSend.setProvider(
+    self.brokerCtx,
+    proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
+      return await self.send(envelope),
+  )
+
+  self.started = true
+  ok()
+
+proc stop*(self: MessagingClient) {.async.} =
+  if not self.started:
+    return
+  MessagingSend.clearProvider(self.brokerCtx)
+  await self.sendService.stopSendService()
+  await self.recvService.stopRecvService()
+  self.started = false

logos_delivery/waku/persistency/backend_thread.nim

@@ -162,13 +162,13 @@ proc registerProviders(backend: KvBackend, ctx: BrokerContext): Result[void, str
 
   return ok()
 
-proc clearProviders(ctx: BrokerContext) =
+proc clearProviders(ctx: BrokerContext) {.async.} =
   KvGet.clearProvider(ctx)
   KvExists.clearProvider(ctx)
   KvScan.clearProvider(ctx)
   KvCount.clearProvider(ctx)
   KvDelete.clearProvider(ctx)
-  PersistEvent.dropAllListeners(ctx)
+  await PersistEvent.dropAllListeners(ctx)
 
 # ── thread proc ─────────────────────────────────────────────────────────
 
@@ -217,7 +217,7 @@ proc storageThreadMain(arg: ptr StorageThreadArg) {.thread.} =
   except CatchableError as e:
     error "storage thread loop crashed", err = e.msg
 
-  clearProviders(arg.ctx)
+  waitFor clearProviders(arg.ctx)
   backend.close()
 
 # ── lifecycle ───────────────────────────────────────────────────────────

logos_delivery/waku/persistency/persistency.nim

@@ -272,7 +272,7 @@ proc hasJob*(p: Persistency, jobId: string): bool {.inline.} =
 proc persist*(t: Job, ops: seq[TxOp]): Future[void] {.async.} =
   ## Emit a batched persist event. The handler treats >1 ops as a single
   ## BEGIN IMMEDIATE/COMMIT transaction (see backend_sqlite.applyOps).
-  await PersistEvent.emit(t.context, PersistEvent(ops: ops))
+  PersistEvent.emit(t.context, PersistEvent(ops: ops))
 
 proc persist*(t: Job, op: TxOp): Future[void] {.async.} =
   await persist(t, @[op])

nimble.lock

@@ -328,8 +328,8 @@
       }
     },
     "brokers": {
-      "version": "#v3.1.1",
-      "vcsRevision": "a7316a35f1b62e3497ae8ee0fc1aace74df0beb2",
+      "version": "#v3.1.4",
+      "vcsRevision": "8ae9e963b0b4478c93e6f888be6a46654da787de",
       "url": "https://github.com/NagyZoltanPeter/nim-brokers.git",
       "downloadMethod": "git",
       "dependencies": [
@@ -341,7 +341,7 @@
         "cbor_serialization"
       ],
       "checksums": {
-        "sha1": "4447d7c1f9da14ae439afb23aee45116ce2ecb40"
+        "sha1": "2949398033c1b3a2586f0afd7e92f46b9a0a412e"
       }
     },
     "stint": {

nix/deps.nix

@@ -159,8 +159,8 @@
 
   brokers = pkgs.fetchgit {
     url = "https://github.com/NagyZoltanPeter/nim-brokers.git";
-    rev = "a7316a35f1b62e3497ae8ee0fc1aace74df0beb2";
-    sha256 = "1990270n88jm0i48g07zr4vq2nn79g7gymf28f3g5ak42g33l7rm";
+    rev = "8ae9e963b0b4478c93e6f888be6a46654da787de";
+    sha256 = "0r3vrc6h9jiwhd3p4sy3j8gx3k5174s77qcp2wrswclvfvbskw5y";
     fetchSubmodules = true;
   };
 
@@ -277,7 +277,6 @@
   };
 
   sds = pkgs.fetchgit {
-    # Keep in sync with the nim-sds pin in nimble.lock.
     url = "https://github.com/logos-messaging/nim-sds.git";
     rev = "b12f5ee07c5b764303b51fb948b32a4ade1de3b5";
     sha256 = "1z8f0v1ww7y6zssdacjxfs6s4862dwckw25df3yn1v0qnz40rpc8";

tests/channels/test_reliable_channel_send_receive.nim

@@ -11,6 +11,7 @@ import logos_delivery
 import logos_delivery/waku/[waku_node, waku_core]
 import logos_delivery/waku/factory/waku_conf
 import logos_delivery/api/events/messaging_client_events as waku_message_events
+import logos_delivery/api/messaging_client_api
 import tools/confutils/cli_args
 
 import logos_delivery/channels/reliable_channel_manager
@@ -187,16 +188,16 @@ suite "Reliable Channel - send state machine":
     setNoopEncryption()
 
     var sendCalls = 0
-    let fakeSend: SendHandler = proc(
-        env: MessageEnvelope
-    ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
-      sendCalls.inc
-      return ok(fakeMsgReqId)
+    MessagingSend.replaceProvider(
+      brokerCtx,
+      proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
+        sendCalls.inc
+        return ok(fakeMsgReqId),
+    ).isOkOr:
+      raiseAssert "replaceProvider failed: " & error
 
     discard manager
-      .createReliableChannel(
-        channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
-      )
+      .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
       .expect("createReliableChannel")
 
     let sentFut = newFuture[RequestId]("channel-sent")
@@ -252,17 +253,17 @@ suite "Reliable Channel - send state machine":
     setNoopEncryption()
 
     var msgReqIds: seq[RequestId]
-    let fakeSend: SendHandler = proc(
-        env: MessageEnvelope
-    ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
-      let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1))
-      msgReqIds.add(id)
-      return ok(id)
+    MessagingSend.replaceProvider(
+      brokerCtx,
+      proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
+        let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1))
+        msgReqIds.add(id)
+        return ok(id),
+    ).isOkOr:
+      raiseAssert "replaceProvider failed: " & error
 
     discard manager
-      .createReliableChannel(
-        channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
-      )
+      .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
       .expect("createReliableChannel")
 
     let sentFut = newFuture[RequestId]("channel-sent")
@@ -354,28 +355,27 @@ suite "Reliable Channel - send state machine":
 
     var msgReqIds: seq[RequestId]
     var sendsReturned = 0
-    let fakeSend: SendHandler = proc(
-        env: MessageEnvelope
-    ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
-      ## 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.
-      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: ""),
-        )
-        await sleepAsync(50.milliseconds)
-      sendsReturned.inc()
-      return ok(id)
+    MessagingSend.replaceProvider(
+      brokerCtx,
+      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.
+        let id = RequestId("race-msg-req-" & $(msgReqIds.len + 1))
+        msgReqIds.add(id)
+        if msgReqIds.len == 2:
+          MessageSentEvent.emit(
+            brokerCtx, MessageSentEvent(requestId: msgReqIds[0], messageHash: "")
+          )
+          await sleepAsync(50.milliseconds)
+        sendsReturned.inc()
+        return ok(id),
+    ).isOkOr:
+      raiseAssert "replaceProvider failed: " & error
 
     discard manager
-      .createReliableChannel(
-        channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
-      )
+      .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
       .expect("createReliableChannel")
 
     var finalisedReqIds: seq[RequestId]
@@ -457,15 +457,15 @@ suite "Reliable Channel - SDS persistence":
 
     setNoopEncryption()
 
-    let fakeSend: SendHandler = proc(
-        env: MessageEnvelope
-    ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
-      return ok(RequestId("persist-msg-req-1"))
+    MessagingSend.replaceProvider(
+      globalBrokerContext(),
+      proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
+        return ok(RequestId("persist-msg-req-1")),
+    ).isOkOr:
+      raiseAssert "replaceProvider failed: " & error
 
     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")
@@ -798,17 +798,17 @@ suite "Reliable Channel - SDS protocol semantics":
     setNoopEncryption()
 
     var capturedWires: seq[seq[byte]]
-    let fakeSend: SendHandler = proc(
-        env: MessageEnvelope
-    ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
-      ## Noop encryption is identity, so the envelope payload IS the SDS wire.
-      capturedWires.add(env.payload)
-      return ok(RequestId("semantics-req-" & $capturedWires.len))
+    MessagingSend.replaceProvider(
+      brokerCtx,
+      proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
+        ## Noop encryption is identity, so the envelope payload IS the SDS wire.
+        capturedWires.add(envelope.payload)
+        return ok(RequestId("semantics-req-" & $capturedWires.len)),
+    ).isOkOr:
+      raiseAssert "replaceProvider failed: " & error
 
     discard manager
-      .createReliableChannel(
-        channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
-      )
+      .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
       .expect("createReliableChannel")
 
     let remotePeer =
@@ -866,16 +866,17 @@ suite "Reliable Channel - SDS protocol semantics":
     setNoopEncryption()
 
     var capturedWires: seq[seq[byte]]
-    let fakeSend: SendHandler = proc(
-        env: MessageEnvelope
-    ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
-      capturedWires.add(env.payload)
-      return ok(RequestId("ack-req-" & $capturedWires.len))
+    MessagingSend.replaceProvider(
+      brokerCtx,
+      proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
+        ## Noop encryption is identity, so the envelope payload IS the SDS wire.
+        capturedWires.add(envelope.payload)
+        return ok(RequestId("ack-req-" & $capturedWires.len)),
+    ).isOkOr:
+      raiseAssert "replaceProvider failed: " & error
 
     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")
@@ -1102,16 +1103,17 @@ suite "Reliable Channel - SDS protocol semantics":
     setNoopEncryption()
 
     var capturedWires: seq[seq[byte]]
-    let fakeSend: SendHandler = proc(
-        env: MessageEnvelope
-    ): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
-      capturedWires.add(env.payload)
-      return ok(RequestId("unique-req-" & $capturedWires.len))
+    MessagingSend.replaceProvider(
+      brokerCtx,
+      proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
+        ## Noop encryption is identity, so the envelope payload IS the SDS wire.
+        capturedWires.add(envelope.payload)
+        return ok(RequestId("unique-req-" & $capturedWires.len)),
+    ).isOkOr:
+      raiseAssert "replaceProvider failed: " & error
 
     discard manager
-      .createReliableChannel(
-        channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
-      )
+      .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
       .expect("createReliableChannel")
 
     var deliveries: seq[seq[byte]]

tests/node/test_wakunode_health_monitor.nim

@@ -29,6 +29,7 @@ import logos_delivery/waku/node/subscription_manager
 import logos_delivery/waku/waku
 import logos_delivery/waku/factory/waku_state_info
 import logos_delivery/messaging/messaging_client
+import logos_delivery/messaging/messaging_client_lifecycle
 
 const MockDLow = 4 # Mocked GossipSub DLow value
 

tests/persistency/test_lifecycle.nim

@@ -156,7 +156,7 @@ suite "Persistency lifecycle":
     let ev = PersistEvent(
       ops: @[TxOp(category: "msg", key: k, kind: txPut, payload: payloadBytes("hello"))]
     )
-    await PersistEvent.emit(t.context, ev)
+    PersistEvent.emit(t.context, ev)
     let ckOk2 = await t.pollExists("msg", k)
     check ckOk2
 
@@ -178,7 +178,7 @@ suite "Persistency lifecycle":
     check a.context != b.context
 
     let k = key("shared", 1'i64)
-    await PersistEvent.emit(
+    PersistEvent.emit(
       a.context,
       PersistEvent(
         ops: @[
@@ -188,7 +188,7 @@ suite "Persistency lifecycle":
         ]
       ),
     )
-    await PersistEvent.emit(
+    PersistEvent.emit(
       b.context,
       PersistEvent(
         ops: @[
@@ -255,7 +255,7 @@ suite "Persistency lifecycle":
       ops.add(
         TxOp(category: "msg", key: key("c", i), kind: txPut, payload: payloadBytes($i))
       )
-    await PersistEvent.emit(t.context, PersistEvent(ops: ops))
+    PersistEvent.emit(t.context, PersistEvent(ops: ops))
     # Wait for the last insert to land.
     let ckOk5 = await t.pollExists("msg", key("c", 5'i64))
     check ckOk5
@@ -285,7 +285,7 @@ suite "Persistency lifecycle":
     let r1 = aw7.get()
     check r1.existed == false
 
-    await PersistEvent.emit(
+    PersistEvent.emit(
       t.context,
       PersistEvent(
         ops: @[TxOp(category: "msg", key: k, kind: txPut, payload: payloadBytes("v"))]