From 2c1ad9f76091e913cda8937097573549be410b25 Mon Sep 17 00:00:00 2001 From: Lorenzo Delgado Date: Fri, 28 Oct 2022 16:30:01 +0200 Subject: [PATCH] refactor(node): use lightpush client in waku_node --- apps/chat2/chat2.nim | 7 ++- apps/wakunode2/wakunode2.nim | 13 ++-- tests/v2/test_waku_lightpush.nim | 6 +- tests/v2/test_wakunode_lightpush.nim | 12 +--- waku/v2/node/waku_node.nim | 62 +++++++++++++------ waku/v2/protocol/waku_lightpush/client.nim | 25 +++----- waku/v2/protocol/waku_lightpush/protocol.nim | 44 +------------ .../waku_lightpush/protocol_metrics.nim | 1 - 8 files changed, 68 insertions(+), 102 deletions(-) diff --git a/apps/chat2/chat2.nim b/apps/chat2/chat2.nim index f1b38acd8..31ee45228 100644 --- a/apps/chat2/chat2.nim +++ b/apps/chat2/chat2.nim @@ -242,7 +242,7 @@ proc publish(c: Chat, line: string) = c.node.wakuRlnRelay.lastEpoch = message.proof.epoch if not c.node.wakuLightPush.isNil(): # Attempt lightpush - asyncSpawn c.node.lightpush2(DefaultTopic, message) + asyncSpawn c.node.lightpushPublish(DefaultTopic, message) else: asyncSpawn c.node.publish(DefaultTopic, message, handler) else: @@ -271,7 +271,7 @@ proc publish(c: Chat, line: string) = if not c.node.wakuLightPush.isNil(): # Attempt lightpush - asyncSpawn c.node.lightpush2(DefaultTopic, message) + asyncSpawn c.node.lightpushPublish(DefaultTopic, message) else: asyncSpawn c.node.publish(DefaultTopic, message) @@ -493,7 +493,8 @@ proc processInput(rfd: AsyncFD) {.async.} = if conf.lightpushnode != "": await mountLightPush(node) - node.wakuLightPush.setPeer(parseRemotePeerInfo(conf.lightpushnode)) + node.mountLightPushClient() + node.setLightPushPeer(conf.lightpushnode) if conf.filternode != "": await node.mountFilter() diff --git a/apps/wakunode2/wakunode2.nim b/apps/wakunode2/wakunode2.nim index 903ac69c4..572e42577 100644 --- a/apps/wakunode2/wakunode2.nim +++ b/apps/wakunode2/wakunode2.nim @@ -400,17 +400,18 @@ proc setupProtocols(node: WakuNode, conf: WakuNodeConf, return err("failed to set node waku store peer: " & getCurrentExceptionMsg()) # NOTE Must be mounted after relay - if (conf.lightpushnode != "") or (conf.lightpush): + if conf.lightpush: try: await mountLightPush(node) except: return err("failed to mount waku lightpush protocol: " & getCurrentExceptionMsg()) - if conf.lightpushnode != "": - try: - setLightPushPeer(node, conf.lightpushnode) - except: - return err("failed to set node waku lightpush peer: " & getCurrentExceptionMsg()) + if conf.lightpushnode != "": + try: + mountLightPushClient(node) + setLightPushPeer(node, conf.lightpushnode) + except: + return err("failed to set node waku lightpush peer: " & getCurrentExceptionMsg()) # Filter setup. NOTE Must be mounted after relay if (conf.filternode != "") or (conf.filter): diff --git a/tests/v2/test_waku_lightpush.nim b/tests/v2/test_waku_lightpush.nim index 34ba8c3d2..61f889b1d 100644 --- a/tests/v2/test_waku_lightpush.nim +++ b/tests/v2/test_waku_lightpush.nim @@ -60,8 +60,7 @@ suite "Waku Lightpush": message = fakeWakuMessage() ## When - let rpc = PushRequest(pubSubTopic: topic, message: message) - let requestRes = await client.request(rpc, serverPeerId) + let requestRes = await client.publish(topic, message, peer=serverPeerId) require await handlerFuture.withTimeout(100.millis) @@ -105,8 +104,7 @@ suite "Waku Lightpush": message = fakeWakuMessage() ## When - let rpc = PushRequest(pubSubTopic: topic, message: message) - let requestRes = await client.request(rpc, serverPeerId) + let requestRes = await client.publish(topic, message, peer=serverPeerId) require await handlerFuture.withTimeout(100.millis) diff --git a/tests/v2/test_wakunode_lightpush.nim b/tests/v2/test_wakunode_lightpush.nim index 1e4ebc54f..fcd569813 100644 --- a/tests/v2/test_wakunode_lightpush.nim +++ b/tests/v2/test_wakunode_lightpush.nim @@ -35,7 +35,7 @@ procSuite "WakuNode - Lightpush": await destNode.mountRelay(@[DefaultPubsubTopic]) await bridgeNode.mountRelay(@[DefaultPubsubTopic]) await bridgeNode.mountLightPush() - await lightNode.mountLightPush() + lightNode.mountLightPushClient() discard await lightNode.peerManager.dialPeer(bridgeNode.peerInfo.toRemotePeerInfo(), WakuLightPushCodec) await sleepAsync(100.milliseconds) @@ -57,16 +57,10 @@ procSuite "WakuNode - Lightpush": await sleepAsync(100.millis) ## When - let lightpushRes = await lightNode.lightpush(DefaultPubsubTopic, message) + await lightNode.lightpushPublish(DefaultPubsubTopic, message) - require await completionFutRelay.withTimeout(5.seconds) - ## Then - check lightpushRes.isOk() - - let response = lightpushRes.get() - check: - response.isSuccess == true + check await completionFutRelay.withTimeout(5.seconds) ## Cleanup await allFutures(lightNode.stop(), bridgeNode.stop(), destNode.stop()) diff --git a/waku/v2/node/waku_node.nim b/waku/v2/node/waku_node.nim index dca9046eb..3358add31 100644 --- a/waku/v2/node/waku_node.nim +++ b/waku/v2/node/waku_node.nim @@ -21,6 +21,7 @@ import ../protocol/waku_swap/waku_swap, ../protocol/waku_filter, ../protocol/waku_lightpush, + ../protocol/waku_lightpush/client, ../protocol/waku_rln_relay/waku_rln_relay_types, ../protocol/waku_peer_exchange, ../utils/[peers, requests, wakuenr], @@ -37,6 +38,8 @@ declarePublicGauge waku_version, "Waku version info (in git describe format)", [ declarePublicCounter waku_node_messages, "number of messages received", ["type"] declarePublicGauge waku_node_filters, "number of content filter subscriptions" declarePublicGauge waku_node_errors, "number of wakunode errors", ["type"] +declarePublicGauge waku_lightpush_peers, "number of lightpush peers" + logScope: topics = "wakunode" @@ -76,6 +79,7 @@ type wakuSwap*: WakuSwap wakuRlnRelay*: WakuRLNRelay wakuLightPush*: WakuLightPush + wakuLightpushClient*: WakuLightPushClient wakuPeerExchange*: WakuPeerExchange enr*: enr.Record libp2pPing*: Ping @@ -589,13 +593,18 @@ proc resume*(node: WakuNode, peerList: Option[seq[RemotePeerInfo]] = none(seq[Re ## Waku lightpush +proc mountLightPushClient*(node: WakuNode) = + info "mounting light push client" + + node.wakuLightpushClient = WakuLightPushClient.new(node.peerManager, node.rng) + + proc mountLightPush*(node: WakuNode) {.async.} = info "mounting light push" var pushHandler: PushMessageHandler if node.wakuRelay.isNil(): debug "mounting lightpush without relay (nil)" - # TODO: Remove after using waku lightpush client pushHandler = proc(peer: PeerId, pubsubTopic: string, message: WakuMessage): Future[WakuLightPushResult[void]] {.async.} = return err("no waku relay found") else: @@ -612,29 +621,46 @@ proc mountLightPush*(node: WakuNode) {.async.} = node.switch.mount(node.wakuLightPush, protocolMatcher(WakuLightPushCodec)) -proc setLightPushPeer*(node: WakuNode, peer: RemotePeerInfo|string) {.raises: [Defect, ValueError, LPError].} = - if node.wakuLightPush.isNil(): - error "could not set peer, waku lightpush is nil" - return - - info "Set lightpush peer", peer=peer - - let remotePeer = when peer is string: parseRemotePeerInfo(peer) - else: peer - node.wakuLightPush.setPeer(remotePeer) - -proc lightpush*(node: WakuNode, topic: PubsubTopic, message: WakuMessage): Future[WakuLightpushResult[PushResponse]] {.async, gcsafe.} = +proc lightpushPublish*(node: WakuNode, pubsubTopic: PubsubTopic, message: WakuMessage, peer: RemotePeerInfo): Future[WakuLightPushResult[void]] {.async, gcsafe.} = ## Pushes a `WakuMessage` to a node which relays it further on PubSub topic. ## Returns whether relaying was successful or not. ## `WakuMessage` should contain a `contentTopic` field for light node ## functionality. - debug "Publishing with lightpush", topic=topic, contentTopic=message.contentTopic + if node.wakuLightpushClient.isNil(): + return err("waku lightpush client is nil") - let rpc = PushRequest(pubSubTopic: topic, message: message) - return await node.wakuLightPush.request(rpc) + debug "publishing message with lightpush", pubsubTopic=pubsubTopic, contentTopic=message.contentTopic, peer=peer -proc lightpush2*(node: WakuNode, topic: PubsubTopic, message: WakuMessage) {.async, gcsafe.} = - discard await node.lightpush(topic, message) + return await node.wakuLightpushClient.publish(pubsubTopic, message, peer) + + +# TODO: Move to application module (e.g., wakunode2.nim) +proc setLightPushPeer*(node: WakuNode, peer: RemotePeerInfo|string) {.raises: [Defect, ValueError, LPError], + deprecated: "Use 'node.lightpushPublish()' instead".} = + debug "seting lightpush client peer", peer=peer + + let remotePeer = when peer is string: parseRemotePeerInfo(peer) + else: peer + node.peerManager.addPeer(remotePeer, WakuLightPushCodec) + waku_lightpush_peers.inc() + +# TODO: Move to application module (e.g., wakunode2.nim) +proc lightpushPublish*(node: WakuNode, pubsubTopic: PubsubTopic, message: WakuMessage): Future[void] {.async, gcsafe, + deprecated: "Use 'node.lightpushPublish()' instead".} = + if node.wakuLightpushClient.isNil(): + error "failed to publish message", error="waku lightpush client is nil" + return + + let peerOpt = node.peerManager.selectPeer(WakuLightPushCodec) + if peerOpt.isNone(): + error "failed to publish message", error="no suitable remote peers" + return + + let publishRes = await node.lightpushPublish(pubsubTopic, message, peer=peerOpt.get()) + if publishRes.isOk(): + return + + error "failed to publish message", error=publishRes.error ## Waku peer-exchange diff --git a/waku/v2/protocol/waku_lightpush/client.nim b/waku/v2/protocol/waku_lightpush/client.nim index 5da0ab1d1..329be7aff 100644 --- a/waku/v2/protocol/waku_lightpush/client.nim +++ b/waku/v2/protocol/waku_lightpush/client.nim @@ -10,6 +10,7 @@ import import ../../node/peer_manager/peer_manager, ../../utils/requests, + ../waku_message, ./protocol, ./protocol_metrics, ./rpc, @@ -31,7 +32,7 @@ proc new*(T: type WakuLightPushClient, WakuLightPushClient(peerManager: peerManager, rng: rng) -proc request*(wl: WakuLightPushClient, req: PushRequest, peer: RemotePeerInfo): Future[WakuLightPushResult[void]] {.async, gcsafe.} = +proc sendPushRequest(wl: WakuLightPushClient, req: PushRequest, peer: PeerId|RemotePeerInfo): Future[WakuLightPushResult[void]] {.async, gcsafe.} = let connOpt = await wl.peerManager.dialPeer(peer, WakuLightPushCodec) if connOpt.isNone(): waku_lightpush_errors.inc(labelValues = [dialFailure]) @@ -41,8 +42,8 @@ proc request*(wl: WakuLightPushClient, req: PushRequest, peer: RemotePeerInfo): let rpc = PushRPC(requestId: generateRequestId(wl.rng), request: req) await connection.writeLP(rpc.encode().buffer) - var message = await connection.readLp(MaxRpcSize.int) - let decodeRespRes = PushRPC.init(message) + var buffer = await connection.readLp(MaxRpcSize.int) + let decodeRespRes = PushRPC.init(buffer) if decodeRespRes.isErr(): error "failed to decode response" waku_lightpush_errors.inc(labelValues = [decodeRpcFailure]) @@ -62,18 +63,6 @@ proc request*(wl: WakuLightPushClient, req: PushRequest, peer: RemotePeerInfo): return ok() - -### Set lightpush peer and send push requests - -proc setPeer*(wl: WakuLightPushClient, peer: RemotePeerInfo) = - wl.peerManager.addPeer(peer, WakuLightPushCodec) - waku_lightpush_peers.inc() - -proc request*(wl: WakuLightPushClient, req: PushRequest): Future[WakuLightPushResult[void]] {.async, gcsafe.} = - let peerOpt = wl.peerManager.selectPeer(WakuLightPushCodec) - if peerOpt.isNone(): - error "no suitable remote peers" - waku_lightpush_errors.inc(labelValues = [peerNotFoundFailure]) - return err(peerNotFoundFailure) - - return await wl.request(req, peerOpt.get()) +proc publish*(wl: WakuLightPushClient, pubsubTopic: string, message: WakuMessage, peer: PeerId|RemotePeerInfo): Future[WakuLightPushResult[void]] {.async, gcsafe.} = + let pushRequest = PushRequest(pubsubTopic: pubsubTopic, message: message) + return await wl.sendPushRequest(pushRequest, peer) diff --git a/waku/v2/protocol/waku_lightpush/protocol.nim b/waku/v2/protocol/waku_lightpush/protocol.nim index ce9678c5a..8fe7282ba 100644 --- a/waku/v2/protocol/waku_lightpush/protocol.nim +++ b/waku/v2/protocol/waku_lightpush/protocol.nim @@ -8,10 +8,8 @@ import metrics, bearssl/rand import - ../waku_message, - ../waku_relay, ../../node/peer_manager/peer_manager, - ../../utils/requests, + ../waku_message, ./rpc, ./rpc_codec, ./protocol_metrics @@ -77,43 +75,3 @@ proc new*(T: type WakuLightPush, let wl = WakuLightPush(rng: rng, peerManager: peerManager, pushHandler: pushHandler) wl.initProtocolHandler() return wl - - -proc setPeer*(wlp: WakuLightPush, peer: RemotePeerInfo) {. - deprecated: "Use 'WakuLightPushClient.setPeer()' instead" .} = - wlp.peerManager.addPeer(peer, WakuLightPushCodec) - waku_lightpush_peers.inc() - -proc request(wl: WakuLightPush, req: PushRequest, peer: RemotePeerInfo): Future[WakuLightPushResult[PushResponse]] {.async, gcsafe, - deprecated: "Use 'WakuLightPushClient.request()' instead" .} = - let connOpt = await wl.peerManager.dialPeer(peer, WakuLightPushCodec) - if connOpt.isNone(): - waku_lightpush_errors.inc(labelValues = [dialFailure]) - return err(dialFailure) - - let connection = connOpt.get() - - let rpc = PushRPC(requestId: generateRequestId(wl.rng), request: req) - await connection.writeLP(rpc.encode().buffer) - - var message = await connection.readLp(MaxRpcSize.int) - let res = PushRPC.init(message) - - if res.isErr(): - waku_lightpush_errors.inc(labelValues = [decodeRpcFailure]) - return err(decodeRpcFailure) - - let rpcRes = res.get() - if rpcRes.response == PushResponse(): - return err("empty response body") - - return ok(rpcRes.response) - -proc request*(wl: WakuLightPush, req: PushRequest): Future[WakuLightPushResult[PushResponse]] {.async, gcsafe, - deprecated: "Use 'WakuLightPushClient.request()' instead" .} = - let peerOpt = wl.peerManager.selectPeer(WakuLightPushCodec) - if peerOpt.isNone(): - waku_lightpush_errors.inc(labelValues = [dialFailure]) - return err(dialFailure) - - return await wl.request(req, peerOpt.get()) diff --git a/waku/v2/protocol/waku_lightpush/protocol_metrics.nim b/waku/v2/protocol/waku_lightpush/protocol_metrics.nim index 469da1c94..b55486378 100644 --- a/waku/v2/protocol/waku_lightpush/protocol_metrics.nim +++ b/waku/v2/protocol/waku_lightpush/protocol_metrics.nim @@ -3,7 +3,6 @@ import metrics -declarePublicGauge waku_lightpush_peers, "number of lightpush peers" declarePublicGauge waku_lightpush_errors, "number of lightpush protocol errors", ["type"] declarePublicGauge waku_lightpush_messages, "number of lightpush messages received", ["type"]