From 84a42a3eb4f684283c5dc6dd7246b903f2af97ea Mon Sep 17 00:00:00 2001 From: NagyZoltanPeter <113987313+NagyZoltanPeter@users.noreply.github.com> Date: Wed, 15 Jul 2026 05:44:33 +0200 Subject: [PATCH] refactor(consumers): reuse envelope.hash in archive/filter/REST/FFI; share filter buffer archive.handleMessage and filter.handleMessage gain envelope-native entry points (reusing the precomputed hash) with thin (topic,msg) compat overloads for tests. Filter pushToPeer takes a shared `ref seq[byte]` so the encoded buffer is not deep-copied per subscribed peer. REST messageCacheHandler and the FFI relay handler adopt the envelope; JsonMessageEvent.new reuses the passed-in hash. Co-Authored-By: Claude Fable 5 --- library/events/json_message_event.nim | 14 ++++++-- library/kernel_api/protocols/relay_api.nim | 4 +-- logos_delivery/waku/rest_api/handlers.nim | 4 +-- logos_delivery/waku/waku_archive/archive.nim | 16 ++++++--- .../waku/waku_filter_v2/protocol.nim | 35 ++++++++++++------- 5 files changed, 50 insertions(+), 23 deletions(-) diff --git a/library/events/json_message_event.nim b/library/events/json_message_event.nim index 61278b4fa..ca44300c3 100644 --- a/library/events/json_message_event.nim +++ b/library/events/json_message_event.nim @@ -69,9 +69,15 @@ type JsonMessageEvent* = ref object of JsonEvent messageHash*: string wakuMessage*: JsonMessage -proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T = +proc new*( + T: type JsonMessageEvent, + pubSubTopic: string, + msg: WakuMessage, + msgHash: WakuMessageHash, +): T = # Returns a WakuMessage event as indicated in # https://github.com/vacp2p/rfc/blob/master/content/docs/rfcs/36/README.md#jsonmessageevent-type + # `msgHash` is the precomputed message hash (reused from the inbound envelope). var payload = newSeq[byte](len(msg.payload)) if len(msg.payload) != 0: @@ -85,8 +91,6 @@ proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T = if len(msg.proof) != 0: copyMem(addr proof[0], unsafeAddr msg.proof[0], len(msg.proof)) - let msgHash = computeMessageHash(pubSubTopic, msg) - return JsonMessageEvent( eventType: "message", pubSubTopic: pubSubTopic, @@ -102,5 +106,9 @@ proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T = ), ) +proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T = + ## Convenience overload computing the hash for callers without one. + JsonMessageEvent.new(pubSubTopic, msg, computeMessageHash(pubSubTopic, msg)) + method `$`*(jsonMessage: JsonMessageEvent): string = $(%*jsonMessage) diff --git a/library/kernel_api/protocols/relay_api.nim b/library/kernel_api/protocols/relay_api.nim index 7a1fe446f..a2e23398c 100644 --- a/library/kernel_api/protocols/relay_api.nim +++ b/library/kernel_api/protocols/relay_api.nim @@ -78,9 +78,9 @@ proc waku_relay_subscribe( pubSubTopic: cstring, ) {.ffi.} = proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): WakuRelayHandler = - return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} = + return proc(envelope: WakuEnvelope) {.async.} = callEventCallback(ctx, "onReceivedMessage"): - $JsonMessageEvent.new(pubsubTopic, msg) + $JsonMessageEvent.new(envelope.pubsubTopic, envelope.msg, envelope.hash) ( await ctx.myLib[].waku.relaySubscribe( diff --git a/logos_delivery/waku/rest_api/handlers.nim b/logos_delivery/waku/rest_api/handlers.nim index 520c0518a..df36bdf3e 100644 --- a/logos_delivery/waku/rest_api/handlers.nim +++ b/logos_delivery/waku/rest_api/handlers.nim @@ -34,5 +34,5 @@ proc defaultDiscoveryHandler*( ### Message Cache proc messageCacheHandler*(cache: MessageCache): WakuRelayHandler = - return proc(pubsubTopic: string, msg: WakuMessage): Future[void] {.async, closure.} = - cache.addMessage(pubsubTopic, msg) + return proc(envelope: WakuEnvelope): Future[void] {.async, closure.} = + cache.addMessage(envelope.pubsubTopic, envelope.msg) diff --git a/logos_delivery/waku/waku_archive/archive.nim b/logos_delivery/waku/waku_archive/archive.nim index ea5e2bf02..cac963922 100644 --- a/logos_delivery/waku/waku_archive/archive.nim +++ b/logos_delivery/waku/waku_archive/archive.nim @@ -95,10 +95,10 @@ proc new*( return ok(archive) -proc handleMessage*( - self: WakuArchive, pubsubTopic: PubsubTopic, msg: WakuMessage -) {.async.} = - let msgHash = computeMessageHash(pubsubTopic, msg) +proc handleMessage*(self: WakuArchive, envelope: WakuEnvelope) {.async.} = + let pubsubTopic = envelope.pubsubTopic + let msg = envelope.msg + let msgHash = envelope.hash let msgHashHex = msgHash.to0xHex() trace "handling message", @@ -144,6 +144,14 @@ proc handleMessage*( timestamp = msg.timestamp, insertDuration = insertDuration +proc handleMessage*( + self: WakuArchive, pubsubTopic: PubsubTopic, msg: WakuMessage +) {.async.} = + ## Convenience overload building the envelope (and its hash) for callers that + ## don't already have one (e.g. tests, REST). The relay dispatch path uses the + ## envelope overload directly to avoid re-hashing. + await self.handleMessage(WakuEnvelope.init(pubsubTopic, msg)) + proc syncMessageIngress*( self: WakuArchive, msgHash: WakuMessageHash, diff --git a/logos_delivery/waku/waku_filter_v2/protocol.nim b/logos_delivery/waku/waku_filter_v2/protocol.nim index ab77ce2dc..7bcd90c42 100644 --- a/logos_delivery/waku/waku_filter_v2/protocol.nim +++ b/logos_delivery/waku/waku_filter_v2/protocol.nim @@ -168,8 +168,10 @@ proc handleSubscribeRequest*( return FilterSubscribeResponse.ok(request.requestId) proc pushToPeer( - wf: WakuFilter, peerId: PeerId, buffer: seq[byte] + wf: WakuFilter, peerId: PeerId, buffer: ref seq[byte] ): Future[Result[void, string]] {.async.} = + ## `buffer` is shared by reference across all target peers (encoded once in + ## `pushToPeers`) so no per-peer async-closure copy of the payload happens. info "pushing message to subscribed peer", peerId = shortLog(peerId) let stream = ( @@ -178,21 +180,21 @@ proc pushToPeer( error "pushToPeer failed", error return err("pushToPeer failed: " & $error) - await stream.writeLp(buffer) + await stream.writeLp(buffer[]) info "published successful", peerId = shortLog(peerId), stream waku_service_network_bytes.inc( - amount = buffer.len().int64, labelValues = [WakuFilterPushCodec, "out"] + amount = buffer[].len().int64, labelValues = [WakuFilterPushCodec, "out"] ) return ok() proc pushToPeers( - wf: WakuFilter, peers: seq[PeerId], messagePush: MessagePush + wf: WakuFilter, peers: seq[PeerId], messagePush: MessagePush, msgHash: string ) {.async.} = + ## `msgHash` is the precomputed 0x-hex hash of the pushed message (reused from + ## the inbound envelope; not recomputed here). let targetPeerIds = peers.mapIt(shortLog(it)) - let msgHash = - messagePush.pubsubTopic.computeMessageHash(messagePush.wakuMessage).to0xHex() ## it's also refresh expire of msghash, that's why update cache every time, even if it has a value. if wf.messageCache.put(msgHash, Moment.now()): @@ -210,7 +212,8 @@ proc pushToPeers( target_peer_ids = targetPeerIds, msg_hash = msgHash - let bufferToPublish = messagePush.encode().buffer + let bufferToPublish = new(seq[byte]) + bufferToPublish[] = messagePush.encode().buffer var pushFuts: seq[Future[Result[void, string]]] for peerId in peers: @@ -240,10 +243,10 @@ proc maintainSubscriptions*(wf: WakuFilter) {.async.} = waku_filter_subscriptions.set(wf.subscriptions.peersSubscribed.len.float64) const MessagePushTimeout = 20.seconds -proc handleMessage*( - wf: WakuFilter, pubsubTopic: PubsubTopic, message: WakuMessage -) {.async.} = - let msgHash = computeMessageHash(pubsubTopic, message).to0xHex() +proc handleMessage*(wf: WakuFilter, envelope: WakuEnvelope) {.async.} = + let pubsubTopic = envelope.pubsubTopic + let message = envelope.msg + let msgHash = envelope.hash.to0xHex() info "handling message", pubsubTopic = pubsubTopic, contentTopic = message.contentTopic, msg_hash = msgHash @@ -263,7 +266,7 @@ proc handleMessage*( let messagePush = MessagePush(pubsubTopic: pubsubTopic, wakuMessage: message) - if not await wf.pushToPeers(subscribedPeers, messagePush).withTimeout( + if not await wf.pushToPeers(subscribedPeers, messagePush, msgHash).withTimeout( MessagePushTimeout ): error "timed out pushing message to peers", @@ -287,6 +290,14 @@ proc handleMessage*( # Duration in seconds with millisecond precision floating point waku_filter_handle_message_duration_seconds.observe(handleMessageDurationSec) +proc handleMessage*( + wf: WakuFilter, pubsubTopic: PubsubTopic, message: WakuMessage +) {.async.} = + ## Convenience overload building the envelope (and its hash) for callers that + ## don't already have one (e.g. tests). The relay dispatch path uses the + ## envelope overload directly to avoid re-hashing. + await wf.handleMessage(WakuEnvelope.init(pubsubTopic, message)) + proc initProtocolHandler(wf: WakuFilter) = proc handler(conn: Connection, proto: string) {.async: (raises: [CancelledError]).} = info "filter subscribe request handler triggered",