messaging: depend on the Waku kernel, not the raw WakuNode (#4000)

logos_delivery/logos_delivery.nim

@@ -25,11 +25,12 @@ import logos_delivery/waku/waku
 export waku
 import
   logos_delivery/waku/api/[
-    topics, relay, filter, lightpush, store, peer_manager, discovery, debug, health,
-    ping,
+    topics, relay, subscriptions, filter, lightpush, store, peer_manager, discovery,
+    debug, health, ping,
   ]
 export
-  topics, relay, filter, lightpush, store, peer_manager, discovery, debug, health, ping
+  topics, relay, subscriptions, filter, lightpush, store, peer_manager, discovery,
+  debug, health, ping
 # `MessageSeenEvent` is surfaced via `export waku` (Kernel interface); the
 # remaining waku health events live here.
 import logos_delivery/waku/api/events/health_events
@@ -97,7 +98,7 @@ proc new*(
   let waku = (await Waku.new(layerConf.waku, appCallbacks)).valueOr:
     return err("failed to create Waku: " & error)
 
-  let messagingClient = MessagingClient.new(layerConf.messaging, waku.node).valueOr:
+  let messagingClient = MessagingClient.new(layerConf.messaging, waku).valueOr:
     return err("failed to create MessagingClient: " & error)
 
   let reliableChannelManager = ReliableChannelManager.new(

logos_delivery/messaging/api/send.nim

@@ -3,7 +3,8 @@ import results, chronos, chronicles
 
 import logos_delivery/api/types
 import logos_delivery/messaging/messaging_client
-import logos_delivery/waku/node/[waku_node, subscription_manager]
+import logos_delivery/waku/waku
+import logos_delivery/waku/api/subscriptions
 import logos_delivery/messaging/delivery_service/send_service
 import logos_delivery/messaging/delivery_service/send_service/delivery_task
 
@@ -16,17 +17,16 @@ proc send*(
   ## id the caller can correlate with `MessageSentEvent` / `MessageErrorEvent`.
   ?self.checkApiAvailability()
 
-  let isSubbed =
-    self.node.subscriptionManager.isSubscribed(envelope.contentTopic).valueOr(false)
+  let isSubbed = self.waku.isSubscribed(envelope.contentTopic).valueOr(false)
   if not isSubbed:
     info "Auto-subscribing to topic on send", contentTopic = envelope.contentTopic
-    self.node.subscriptionManager.subscribe(envelope.contentTopic).isOkOr:
+    self.waku.subscribe(envelope.contentTopic).isOkOr:
       warn "Failed to auto-subscribe", error = error
       return err("Failed to auto-subscribe before sending: " & error)
 
-  let requestId = RequestId.new(self.node.rng)
+  let requestId = RequestId.new(self.waku.rng)
 
-  let deliveryTask = DeliveryTask.new(requestId, envelope, self.node.brokerCtx).valueOr:
+  let deliveryTask = DeliveryTask.new(requestId, envelope, self.waku.brokerCtx).valueOr:
     return err("MessagingClient.send: Failed to create delivery task: " & error)
 
   asyncSpawn self.sendService.send(deliveryTask)

logos_delivery/messaging/api/subscription.nim

@@ -3,16 +3,17 @@ import results, chronos
 
 import logos_delivery/api/types
 import logos_delivery/messaging/messaging_client
-import logos_delivery/waku/node/[waku_node, subscription_manager]
+import logos_delivery/waku/waku
+import logos_delivery/waku/api/subscriptions
 
 proc subscribe*(
     self: MessagingClient, contentTopic: ContentTopic
 ): Future[Result[void, string]] {.async.} =
   ?self.checkApiAvailability()
-  return self.node.subscriptionManager.subscribe(contentTopic)
+  return self.waku.subscribe(contentTopic)
 
 proc unsubscribe*(
     self: MessagingClient, contentTopic: ContentTopic
 ): Result[void, string] =
   ?self.checkApiAvailability()
-  return self.node.subscriptionManager.unsubscribe(contentTopic)
+  return self.waku.unsubscribe(contentTopic)

logos_delivery/messaging/delivery_service/recv_service/recv_service.nim

@@ -7,15 +7,9 @@ import std/[tables, sequtils, options, sets]
 import chronos, chronicles, libp2p/utility
 import brokers/broker_context
 import
-  logos_delivery/waku/[
-    waku_core,
-    waku_core/topics,
-    waku_store/client,
-    waku_store/common,
-    waku_filter_v2/client,
-    waku_node,
-    node/subscription_manager,
-  ]
+  logos_delivery/waku/[waku_core, waku_core/topics, waku_store/common],
+  logos_delivery/waku/waku,
+  logos_delivery/waku/api/[store, subscriptions]
 import
   logos_delivery/api/kernel_api, # MessageSeenEvent
   logos_delivery/api/messaging_client_api, # MessageReceivedEvent
@@ -38,7 +32,7 @@ type RecvMessage = object
 
 type RecvService* = ref object of RootObj
   brokerCtx: BrokerContext
-  node: WakuNode
+  waku: Waku
   seenMsgListener: MessageSeenEventListener
   connStatusListener: EventConnectionStatusChangeListener
 
@@ -60,7 +54,7 @@ proc getMissingMsgsFromStore(
     self: RecvService, msgHashes: seq[WakuMessageHash]
 ): Future[Result[seq[TupleHashAndMsg], string]] {.async.} =
   let storeResp: StoreQueryResponse = (
-    await self.node.wakuStoreClient.queryToAny(
+    await self.waku.storeQueryToAny(
       StoreQueryRequest(includeData: true, messageHashes: msgHashes)
     )
   ).valueOr:
@@ -85,9 +79,7 @@ proc processIncomingMessage(
   ## or if the message is a duplicate (recently-seen). Otherwise, save it as
   ## recently-seen, emit a MessageReceivedEvent, and return true.
 
-  if not self.node.subscriptionManager.isContentSubscribed(
-    pubsubTopic, message.contentTopic
-  ):
+  if not self.waku.isContentSubscribed(pubsubTopic, message.contentTopic):
     trace "skipping message as I am not subscribed",
       shard = pubsubTopic, contentTopic = message.contentTopic
     return false
@@ -108,16 +100,16 @@ proc processIncomingMessage(
 proc checkStore*(self: RecvService) {.async.} =
   ## Checks the store for messages that were not received directly and
   ## delivers them via MessageReceivedEvent.
-  if self.node.wakuStoreClient.isNil():
+  if not self.waku.isStoreMounted():
     error "recv service has no store client mounted, skipping store check"
     return
 
   self.endTimeToCheck = getNowInNanosecondTime()
 
   ## query store and deliver new recovered messages per subscribed topic
-  for pubsubTopic, contentTopics in self.node.subscriptionManager.subscribedContentTopics:
+  for (pubsubTopic, contentTopics) in self.waku.subscribedContentTopics():
     let storeResp: StoreQueryResponse = (
-      await self.node.wakuStoreClient.queryToAny(
+      await self.waku.storeQueryToAny(
         StoreQueryRequest(
           includeData: false,
           pubsubTopic: some(pubsubTopic),
@@ -171,14 +163,14 @@ proc onConnectionStatusChange(self: RecvService, status: ConnectionStatus) =
     info "recv service backfilling missed messages after coming back online"
     self.backfillHandler = self.checkStore()
 
-proc new*(T: typedesc[RecvService], node: WakuNode): T =
+proc new*(T: typedesc[RecvService], waku: Waku): T =
   ## The storeClient will help to acquire any possible missed messages
 
   let now = getNowInNanosecondTime()
   var recvService = RecvService(
-    node: node,
+    waku: waku,
     startTimeToCheck: now,
-    brokerCtx: node.brokerCtx,
+    brokerCtx: waku.brokerCtx,
     recentReceivedMsgs: @[],
   )
 

logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim

@@ -2,10 +2,8 @@ import logos_delivery/waku/compat/option_valueor
 import chronicles, chronos, results
 import std/options
 import brokers/broker_context
-import
-  logos_delivery/waku/node/peer_manager,
-  logos_delivery/waku/waku_core,
-  logos_delivery/waku/waku_lightpush/[common, client, rpc]
+import logos_delivery/waku/waku_core, logos_delivery/waku/waku
+import logos_delivery/waku/api/publish
 
 import ./[delivery_task, send_processor]
 
@@ -13,27 +11,17 @@ logScope:
   topics = "send service lightpush processor"
 
 type LightpushSendProcessor* = ref object of BaseSendProcessor
-  peerManager: PeerManager
-  lightpushClient: WakuLightPushClient
+  waku: Waku
 
 proc new*(
-    T: typedesc[LightpushSendProcessor],
-    peerManager: PeerManager,
-    lightpushClient: WakuLightPushClient,
-    brokerCtx: BrokerContext,
+    T: typedesc[LightpushSendProcessor], waku: Waku, brokerCtx: BrokerContext
 ): T =
-  return
-    T(peerManager: peerManager, lightpushClient: lightpushClient, brokerCtx: brokerCtx)
-
-proc isLightpushPeerAvailable(
-    self: LightpushSendProcessor, pubsubTopic: PubsubTopic
-): bool =
-  return self.peerManager.selectPeer(WakuLightPushCodec, some(pubsubTopic)).isSome()
+  return T(waku: waku, brokerCtx: brokerCtx)
 
 method isValidProcessor*(
     self: LightpushSendProcessor, task: DeliveryTask
 ): bool {.gcsafe.} =
-  return self.isLightpushPeerAvailable(task.pubsubTopic)
+  return self.waku.lightpushPeerAvailable(task.pubsubTopic)
 
 method sendImpl*(
     self: LightpushSendProcessor, task: DeliveryTask
@@ -44,14 +32,8 @@ method sendImpl*(
     msgHash = task.msgHash.to0xHex(),
     tryCount = task.tryCount
 
-  let peer = self.peerManager.selectPeer(WakuLightPushCodec, some(task.pubsubTopic)).valueOr:
-    debug "No peer available for Lightpush, request pushed back for next round",
-      requestId = task.requestId
-    task.state = DeliveryState.NextRoundRetry
-    return
-
   let numLightpushServers = (
-    await self.lightpushClient.publish(some(task.pubsubTopic), task.msg, peer)
+    await self.waku.lightpushPublishToAny(task.pubsubTopic, task.msg)
   ).valueOr:
     error "LightpushSendProcessor.sendImpl failed", error = error.desc.get($error.code)
     case error.code

logos_delivery/messaging/delivery_service/send_service/send_service.nim

@@ -7,18 +7,9 @@ import chronos, chronicles, libp2p/utility
 import brokers/broker_context
 import
   ./[send_processor, relay_processor, lightpush_processor, delivery_task],
-  logos_delivery/waku/[
-    waku_core,
-    node/waku_node,
-    node/subscription_manager,
-    node/peer_manager,
-    waku_store/client,
-    waku_store/common,
-    waku_relay/protocol,
-    rln/rln,
-    waku_lightpush/client,
-    waku_lightpush/callbacks,
-  ]
+  logos_delivery/waku/[waku_core, waku_store/common],
+  logos_delivery/waku/waku,
+  logos_delivery/waku/api/[store, subscriptions, publish]
 import logos_delivery/api/messaging_client_api
 
 logScope:
@@ -57,19 +48,15 @@ type SendService* = ref object of RootObj
   serviceLoopHandle: Future[void] ## handle that allows to stop the async task
   sendProcessor: BaseSendProcessor
 
-  node: WakuNode
+  waku: Waku
   checkStoreForMessages: bool
   lastStoreCheckTime: Moment ## throttles store validation queries to ArchiveTime cadence
 
 proc setupSendProcessorChain(
-    peerManager: PeerManager,
-    lightpushClient: WakuLightPushClient,
-    relay: WakuRelay,
-    rlnRelay: Rln,
-    brokerCtx: BrokerContext,
+    waku: Waku, brokerCtx: BrokerContext
 ): Result[BaseSendProcessor, string] =
-  let isRelayAvail = not relay.isNil()
-  let isLightPushAvail = not lightpushClient.isNil()
+  let isRelayAvail = waku.hasRelay()
+  let isLightPushAvail = waku.hasLightpush()
 
   if not isRelayAvail and not isLightPushAvail:
     return err("No valid send processor found for the delivery task")
@@ -77,16 +64,10 @@ proc setupSendProcessorChain(
   var processors = newSeq[BaseSendProcessor]()
 
   if isRelayAvail:
-    let rln: Option[Rln] =
-      if rlnRelay.isNil():
-        none[Rln]()
-      else:
-        some(rlnRelay)
-    let publishProc = getRelayPushHandler(relay, rln)
-
+    let publishProc = waku.relayPushHandler()
     processors.add(RelaySendProcessor.new(isLightPushAvail, publishProc, brokerCtx))
   if isLightPushAvail:
-    processors.add(LightpushSendProcessor.new(peerManager, lightpushClient, brokerCtx))
+    processors.add(LightpushSendProcessor.new(waku, brokerCtx))
 
   var currentProcessor: BaseSendProcessor = processors[0]
   for i in 1 ..< processors.len:
@@ -97,26 +78,24 @@ proc setupSendProcessorChain(
   return ok(processors[0])
 
 proc new*(
-    T: typedesc[SendService], preferP2PReliability: bool, w: WakuNode
+    T: typedesc[SendService], preferP2PReliability: bool, waku: Waku
 ): Result[T, string] =
-  if w.wakuRelay.isNil() and w.wakuLightpushClient.isNil():
+  if not waku.hasRelay() and not waku.hasLightpush():
     return err(
       "Could not create SendService. wakuRelay or wakuLightpushClient should be set"
     )
 
-  let checkStoreForMessages = preferP2PReliability and not w.wakuStoreClient.isNil()
+  let checkStoreForMessages = preferP2PReliability and waku.isStoreMounted()
 
-  let sendProcessorChain = setupSendProcessorChain(
-    w.peerManager, w.wakuLightPushClient, w.wakuRelay, w.rln, w.brokerCtx
-  ).valueOr:
+  let sendProcessorChain = setupSendProcessorChain(waku, waku.brokerCtx).valueOr:
     return err("failed to setup SendProcessorChain: " & $error)
 
   let sendService = SendService(
-    brokerCtx: w.brokerCtx,
+    brokerCtx: waku.brokerCtx,
     taskCache: newSeq[DeliveryTask](),
     serviceLoopHandle: nil,
     sendProcessor: sendProcessorChain,
-    node: w,
+    waku: waku,
     checkStoreForMessages: checkStoreForMessages,
     lastStoreCheckTime: Moment.now(),
   )
@@ -127,7 +106,7 @@ proc addTask(self: SendService, task: DeliveryTask) =
   self.taskCache.addUnique(task)
 
 proc isStorePeerAvailable*(sendService: SendService): bool =
-  return sendService.node.peerManager.selectPeer(WakuStoreCodec).isSome()
+  return sendService.waku.hasStorePeer()
 
 proc checkMsgsInStore(self: SendService, tasksToValidate: seq[DeliveryTask]) {.async.} =
   if tasksToValidate.len() == 0:
@@ -142,7 +121,7 @@ proc checkMsgsInStore(self: SendService, tasksToValidate: seq[DeliveryTask]) {.a
   # TODO: confirm hash format for store query!!!
 
   let storeResp: StoreQueryResponse = (
-    await self.node.wakuStoreClient.queryToAny(
+    await self.waku.storeQueryToAny(
       StoreQueryRequest(includeData: false, messageHashes: hashesToValidate)
     )
   ).valueOr:
@@ -292,7 +271,7 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} =
   info "SendService.send: processing delivery task",
     requestId = task.requestId, msgHash = task.msgHash.to0xHex()
 
-  self.node.subscriptionManager.subscribe(task.msg.contentTopic).isOkOr:
+  self.waku.subscribe(task.msg.contentTopic).isOkOr:
     error "SendService.send: failed to subscribe to content topic",
       contentTopic = task.msg.contentTopic, error = error
 

logos_delivery/messaging/messaging_client.nim

@@ -4,7 +4,7 @@
 import results, chronos
 import
   logos_delivery/api/messaging_client_api,
-  logos_delivery/waku/node/waku_node,
+  logos_delivery/waku/waku,
   logos_delivery/messaging/delivery_service/[recv_service, send_service]
 
 # Surfaces the messaging API interface (and its Message* events) to consumers.
@@ -18,19 +18,19 @@ type
     useP2PReliability*: bool
 
   MessagingClient* = ref object of IMessagingClient
-    node*: WakuNode ## Waku core driven by this layer; read by `messaging/api.nim`.
+    waku*: Waku ## The Waku kernel this layer drives; read by `messaging/api/*`.
     sendService*: SendService
     recvService*: RecvService
     started: bool
 
 proc new*(
-    T: type MessagingClient, conf: MessagingClientConf, node: WakuNode
+    T: type MessagingClient, conf: MessagingClientConf, waku: Waku
 ): Result[T, string] =
-  ## The messaging layer chains onto Waku: it drives the underlying
-  ## `WakuNode` (Waku's core) for transport while exposing its own send/recv API.
-  let sendService = ?SendService.new(conf.useP2PReliability, node)
-  let recvService = RecvService.new(node)
-  ok(T(node: node, sendService: sendService, recvService: recvService))
+  ## The messaging layer chains onto Waku: it drives the underlying Waku kernel
+  ## 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:

logos_delivery/waku/api/publish.nim

@@ -0,0 +1,64 @@
+## Waku layer API — message publish primitives used by the messaging send
+## pipeline.
+##
+## Unlike `relay.nim`/`lightpush.nim`, these preserve the rich
+## `WakuLightPushResult` (status code + description) that the send processors
+## branch on for their retry decisions, and expose relay/lightpush availability
+## so the messaging layer never inspects `waku.node` directly.
+import logos_delivery/waku/compat/option_valueor
+{.push raises: [].}
+
+import std/options
+import results, chronos
+
+import logos_delivery/waku/waku
+import
+  logos_delivery/waku/[
+    waku_core,
+    node/waku_node,
+    node/peer_manager,
+    waku_relay/protocol,
+    rln,
+    waku_lightpush/common,
+    waku_lightpush/rpc,
+    waku_lightpush/client,
+    waku_lightpush/callbacks,
+  ]
+
+# WakuLightPushResult, PushMessageHandler, LightPushErrorCode (common) plus the
+# LightPushStatusCode `$`/`==` the send processors branch on (rpc).
+export common, rpc
+
+proc hasRelay*(self: Waku): bool =
+  ## True if relay (gossipsub publishing) is mounted.
+  return not self.node.wakuRelay.isNil()
+
+proc hasLightpush*(self: Waku): bool =
+  ## True if a lightpush client is mounted.
+  return not self.node.wakuLightpushClient.isNil()
+
+proc relayPushHandler*(self: Waku): PushMessageHandler =
+  ## Builds the relay publish handler (appending an RLN proof when RLN is
+  ## mounted) used by the send pipeline. Caller ensures relay is mounted.
+  let rln =
+    if self.node.rln.isNil():
+      none[Rln]()
+    else:
+      some(self.node.rln)
+  return getRelayPushHandler(self.node.wakuRelay, rln)
+
+proc lightpushPeerAvailable*(self: Waku, shard: PubsubTopic): bool =
+  ## True if a lightpush service peer is available for `shard`.
+  return self.node.peerManager.selectPeer(WakuLightPushCodec, some(shard)).isSome()
+
+proc lightpushPublishToAny*(
+    self: Waku, shard: PubsubTopic, message: WakuMessage
+): Future[WakuLightPushResult] {.async.} =
+  ## Selects a lightpush service peer for `shard` and publishes `message`.
+  ## Returns SERVICE_NOT_AVAILABLE when no peer is available.
+  let peer = self.node.peerManager.selectPeer(WakuLightPushCodec, some(shard)).valueOr:
+    return lightpushResultServiceUnavailable("no lightpush peer available for shard")
+  try:
+    return await self.node.wakuLightpushClient.publish(some(shard), message, peer)
+  except CatchableError as e:
+    return lightpushResultInternalError(e.msg)

logos_delivery/waku/api/store.nim

@@ -1,11 +1,36 @@
 ## Waku layer API — store (historical query) operations.
 {.push raises: [].}
 
+import std/options
 import results, chronos, chronicles
 
 import logos_delivery/waku/waku
 import
-  logos_delivery/waku/[waku_core, node/waku_node, waku_store/common, waku_store/client]
+  logos_delivery/waku/
+    [waku_core, node/waku_node, node/peer_manager, waku_store/common, waku_store/client]
+
+proc isStoreMounted*(self: Waku): bool =
+  ## True if a store client is mounted (the node can run store queries).
+  return not self.node.wakuStoreClient.isNil()
+
+proc hasStorePeer*(self: Waku): bool =
+  ## True if at least one store service peer is available to query.
+  return self.node.peerManager.selectPeer(WakuStoreCodec).isSome()
+
+proc storeQueryToAny*(
+    self: Waku, request: StoreQueryRequest
+): Future[Result[StoreQueryResponse, string]] {.async.} =
+  ## Runs a store query against any available store peer (retries across peers).
+  try:
+    if self.node.wakuStoreClient.isNil():
+      return err("wakuStoreClient is not mounted")
+
+    let queryResponse = (await self.node.wakuStoreClient.queryToAny(request)).valueOr:
+      return err($error)
+
+    return ok(queryResponse)
+  except CatchableError as e:
+    return err(e.msg)
 
 proc storeQuery*(
     self: Waku, request: StoreQueryRequest, peer: string, timeoutMs: int

logos_delivery/waku/api/subscriptions.nim

@@ -0,0 +1,37 @@
+## Waku layer API — content-topic subscription operations.
+##
+## These wrap the node's `SubscriptionManager`, which resolves each content
+## topic to its autosharding shard. They give the layers above (messaging) a
+## kernel-level entry point so they never reach into `waku.node` internals.
+{.push raises: [].}
+
+import std/sets
+import results
+
+import logos_delivery/waku/waku
+import logos_delivery/waku/[waku_core, node/waku_node, node/subscription_manager]
+
+proc subscribe*(self: Waku, contentTopic: ContentTopic): Result[void, string] =
+  ## Subscribes to `contentTopic`, resolving its shard via autosharding.
+  return self.node.subscriptionManager.subscribe(contentTopic)
+
+proc unsubscribe*(self: Waku, contentTopic: ContentTopic): Result[void, string] =
+  ## Unsubscribes from `contentTopic`, resolving its shard via autosharding.
+  return self.node.subscriptionManager.unsubscribe(contentTopic)
+
+proc isSubscribed*(self: Waku, contentTopic: ContentTopic): Result[bool, string] =
+  ## True if the node already subscribes to `contentTopic`.
+  return self.node.subscriptionManager.isSubscribed(contentTopic)
+
+proc isContentSubscribed*(
+    self: Waku, shard: PubsubTopic, contentTopic: ContentTopic
+): bool =
+  ## True if `contentTopic` is subscribed on the given `shard` (pubsub topic).
+  return self.node.subscriptionManager.isContentSubscribed(shard, contentTopic)
+
+proc subscribedContentTopics*(self: Waku): seq[(PubsubTopic, HashSet[ContentTopic])] =
+  ## Snapshot of every shard with its non-empty content-topic set.
+  var res: seq[(PubsubTopic, HashSet[ContentTopic])]
+  for shard, contentTopics in self.node.subscriptionManager.subscribedContentTopics:
+    res.add((shard, contentTopics))
+  return res

tests/node/test_wakunode_health_monitor.nim

@@ -26,6 +26,8 @@ import
 
 import ../testlib/[wakunode, wakucore], ../waku_archive/archive_utils
 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
 
 const MockDLow = 4 # Mocked GossipSub DLow value
@@ -228,8 +230,14 @@ suite "Health Monitor - events":
       nodeA.mountMetadata(1, @[0'u16]).expect("Node A failed to mount metadata")
       await nodeA.start()
 
+    # MessagingClient now depends on the Waku kernel, not the raw node. Only
+    # `waku.node` is read on the messaging path; `conf`/`stateInfo` are supplied
+    # solely to satisfy Waku's {.requiresInit.} fields.
+    let waku = Waku(
+      node: nodeA, conf: defaultTestWakuConf(), stateInfo: WakuStateInfo.init(nodeA)
+    )
     let ds = MessagingClient
-      .new(MessagingClientConf(useP2PReliability: false), nodeA)
+      .new(MessagingClientConf(useP2PReliability: false), waku)
       .expect("Failed to create MessagingClient")
     ds.start().expect("Failed to start MessagingClient")
 
@@ -333,8 +341,14 @@ suite "Health Monitor - events":
       nodeA.mountMetadata(1, @[0'u16]).expect("Node A failed to mount metadata")
       await nodeA.start()
 
+    # MessagingClient now depends on the Waku kernel, not the raw node. Only
+    # `waku.node` is read on the messaging path; `conf`/`stateInfo` are supplied
+    # solely to satisfy Waku's {.requiresInit.} fields.
+    let waku = Waku(
+      node: nodeA, conf: defaultTestWakuConf(), stateInfo: WakuStateInfo.init(nodeA)
+    )
     let ds = MessagingClient
-      .new(MessagingClientConf(useP2PReliability: false), nodeA)
+      .new(MessagingClientConf(useP2PReliability: false), waku)
       .expect("Failed to create MessagingClient")
     ds.start().expect("Failed to start MessagingClient")
     let subMgr = nodeA.subscriptionManager