diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim index 435bb968d..d0fb33b90 100644 --- a/logos_delivery/logos_delivery.nim +++ b/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( diff --git a/logos_delivery/messaging/api/send.nim b/logos_delivery/messaging/api/send.nim index cc6312471..27c86d0ab 100644 --- a/logos_delivery/messaging/api/send.nim +++ b/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) diff --git a/logos_delivery/messaging/api/subscription.nim b/logos_delivery/messaging/api/subscription.nim index 35b8c7e53..8893af7f1 100644 --- a/logos_delivery/messaging/api/subscription.nim +++ b/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) diff --git a/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim b/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim index 25b35222d..1090be223 100644 --- a/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim +++ b/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: @[], ) diff --git a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim index ae5e775aa..fa199774b 100644 --- a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim +++ b/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 diff --git a/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim index f5d1f302b..f482a2aff 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/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 diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim index 69c6c520b..4e02c7a82 100644 --- a/logos_delivery/messaging/messaging_client.nim +++ b/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: diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim new file mode 100644 index 000000000..51259504e --- /dev/null +++ b/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) diff --git a/logos_delivery/waku/api/store.nim b/logos_delivery/waku/api/store.nim index 5095dbb0d..a54ebe1f3 100644 --- a/logos_delivery/waku/api/store.nim +++ b/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 diff --git a/logos_delivery/waku/api/subscriptions.nim b/logos_delivery/waku/api/subscriptions.nim new file mode 100644 index 000000000..7638abf3e --- /dev/null +++ b/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 diff --git a/tests/node/test_wakunode_health_monitor.nim b/tests/node/test_wakunode_health_monitor.nim index acf0bc09a..148c93330 100644 --- a/tests/node/test_wakunode_health_monitor.nim +++ b/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