fix: stop emitting receive events after channel_close (#4075)

logos_delivery/channels/reliable_channel.nim

@@ -78,6 +78,10 @@ type
 
     channelReqs: ChannelReqs
     brokerCtx: BrokerContext
+    receivedListener: MessageReceivedEventListener
+    sentListener: MessageSentEventListener
+    errorListener: MessageErrorEventListener
+    closed: bool
 
 func init(
     T: type ChannelReqState,
@@ -102,7 +106,12 @@ func getSenderId*(self: ReliableChannel): SdsParticipantID {.inline.} =
   self.senderId
 
 proc stop*(self: ReliableChannel) {.async: (raises: []).} =
-  ## Stops the SDS background loops. Persisted SDS state survives.
+  ## Drops the event listeners and stops the SDS loops. Persisted SDS state survives.
+  ## `closed` gates any in-flight receive handler that survives the listener drop.
+  self.closed = true
+  await MessageReceivedEvent.dropListener(self.brokerCtx, self.receivedListener)
+  await MessageSentEvent.dropListener(self.brokerCtx, self.sentListener)
+  await MessageErrorEvent.dropListener(self.brokerCtx, self.errorListener)
   await self.sdsHandler.stop()
 
 proc tryFinalizeChannelReq(self: ReliableChannel, channelReqId: RequestId) =
@@ -248,6 +257,8 @@ proc send*(
 
 proc reportReceived(self: ReliableChannel, deliverable: SdsDeliverable) =
   ## Tail of the ingress pipeline (reassemble -> emit).
+  if self.closed:
+    return
   let reassembled = self.segmentation.handleIncomingSegment(deliverable.content)
   if reassembled.isSome():
     ## Emit on the captured `brokerCtx` (the manager's), so the
@@ -292,6 +303,8 @@ proc onMessageReceived(
   ## Invoked from this channel's `MessageReceivedEvent` listener, which
   ## already filtered on the spec marker and on `contentTopic`. The
   ## channel only sees the raw payload bytes for itself.
+  if self.closed:
+    return
 
   ## Notice that the following "request" is implemented implicitly as a broker call to
   ## the `Decrypt` request broker.
@@ -361,7 +374,7 @@ proc new*(
   ## Keeping the listeners (and the handler procs they call) inside the
   ## channel lets `onMessageReceived` / `onMessageFinal` stay private —
   ## the manager doesn't need to know about them.
-  discard MessageReceivedEvent.listen(
+  chn.receivedListener = MessageReceivedEvent.listen(
     chn.brokerCtx,
     proc(evt: MessageReceivedEvent): Future[void] {.async: (raises: []).} =
       ## Drop foreign traffic (non-Reliable-Channel `meta`) and traffic
@@ -372,22 +385,28 @@ proc new*(
         return
       await chn.onMessageReceived(evt.messageHash, evt.message.payload)
     ,
-  )
+  ).valueOr:
+    error "MessageReceivedEvent.listen failed", channelId = channelId, error = error
+    MessageReceivedEventListener()
 
   ## Send-completion events are tagged with the per-segment messaging
   ## `requestId` — globally unique, so we don't need any channel filter
   ## up front. The handler scans this channel's pending entries for a
   ## match and is a no-op when the id belongs to a different channel.
-  discard MessageSentEvent.listen(
+  chn.sentListener = MessageSentEvent.listen(
     chn.brokerCtx,
     proc(evt: MessageSentEvent): Future[void] {.async: (raises: []).} =
       chn.onMessageFinal(evt.requestId, MessagingOutcome.Sent),
-  )
+  ).valueOr:
+    error "MessageSentEvent.listen failed", channelId = channelId, error = error
+    MessageSentEventListener()
 
-  discard MessageErrorEvent.listen(
+  chn.errorListener = MessageErrorEvent.listen(
     chn.brokerCtx,
     proc(evt: MessageErrorEvent): Future[void] {.async: (raises: []).} =
       chn.onMessageFinal(evt.requestId, MessagingOutcome.Failed),
-  )
+  ).valueOr:
+    error "MessageErrorEvent.listen failed", channelId = channelId, error = error
+    MessageErrorEventListener()
 
   return chn

tests/channels/test_reliable_channel_send_receive.nim

@@ -166,6 +166,148 @@ suite "Reliable Channel - ingress":
 
     (await waku.stop()).expect("stop")
 
+  asyncTest "closed channel emits no channel event":
+    ## Issue #4065: no ChannelMessageReceivedEvent after closeChannel.
+    const
+      channelId = ChannelId("test-channel-3")
+      contentTopic = ContentTopic("/reliable-channel/test/proto")
+    let appPayload = "after close".toBytes()
+
+    var waku: LogosDelivery
+    var manager: ReliableChannelManager
+    var brokerCtx: BrokerContext
+    lockNewGlobalBrokerContext:
+      brokerCtx = globalBrokerContext()
+      waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
+      manager = waku.reliableChannelManager
+
+    setNoopEncryption()
+
+    discard manager
+      .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
+      .expect("createReliableChannel")
+
+    (await manager.closeChannel(channelId)).expect("closeChannel")
+
+    var fired = false
+    discard ChannelMessageReceivedEvent
+      .listen(
+        brokerCtx,
+        proc(evt: ChannelMessageReceivedEvent) {.async: (raises: []).} =
+          if evt.channelId == channelId:
+            fired = true
+        ,
+      )
+      .expect("listen ChannelMessageReceivedEvent")
+
+    let remotePeer =
+      ReliabilityManager.new(SdsParticipantID("remote"), ReliabilityConfig.init())
+    let sdsWire = (
+      await remotePeer.wrapOutgoingMessage(
+        appPayload, "close-test-msg-1", SdsChannelID(channelId)
+      )
+    ).expect("wrapOutgoingMessage")
+
+    let inboundMsg = WakuMessage(
+      payload: sdsWire,
+      contentTopic: contentTopic,
+      version: 0,
+      meta: LipWireReliableChannelVersion.toBytes(),
+    )
+
+    waku_message_events.MessageReceivedEvent.emit(
+      brokerCtx,
+      waku_message_events.MessageReceivedEvent(messageHash: "", message: inboundMsg),
+    )
+
+    await sleepAsync(100.milliseconds)
+    check not fired
+
+    (await waku.stop()).expect("stop")
+
+  asyncTest "in-flight receive does not emit after close":
+    ## A message already inside the ingress pipeline when closeChannel
+    ## runs must not emit after close returns.
+    const
+      channelId = ChannelId("test-channel-4")
+      contentTopic = ContentTopic("/reliable-channel/test/proto")
+    let appPayload = "in flight".toBytes()
+
+    var waku: LogosDelivery
+    var manager: ReliableChannelManager
+    var brokerCtx: BrokerContext
+    lockNewGlobalBrokerContext:
+      brokerCtx = globalBrokerContext()
+      waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
+      manager = waku.reliableChannelManager
+
+    setNoopEncryption()
+
+    ## Gate decryption so the receive handler parks mid-pipeline.
+    let decryptGate = newFuture[void]("decrypt-gate")
+    Decrypt
+      .replaceProvider(
+        DefaultBrokerContext,
+        proc(payload: seq[byte]): Future[Result[Decrypt, string]] {.async.} =
+          await decryptGate
+          return ok(Decrypt(payload)),
+      )
+      .expect("replaceProvider")
+
+    discard manager
+      .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
+      .expect("createReliableChannel")
+
+    var fired = false
+    discard ChannelMessageReceivedEvent
+      .listen(
+        brokerCtx,
+        proc(evt: ChannelMessageReceivedEvent) {.async: (raises: []).} =
+          if evt.channelId == channelId:
+            fired = true
+        ,
+      )
+      .expect("listen ChannelMessageReceivedEvent")
+
+    let remotePeer =
+      ReliabilityManager.new(SdsParticipantID("remote"), ReliabilityConfig.init())
+    let sdsWire = (
+      await remotePeer.wrapOutgoingMessage(
+        appPayload, "inflight-test-msg-1", SdsChannelID(channelId)
+      )
+    ).expect("wrapOutgoingMessage")
+
+    let inboundMsg = WakuMessage(
+      payload: sdsWire,
+      contentTopic: contentTopic,
+      version: 0,
+      meta: LipWireReliableChannelVersion.toBytes(),
+    )
+
+    waku_message_events.MessageReceivedEvent.emit(
+      brokerCtx,
+      waku_message_events.MessageReceivedEvent(messageHash: "", message: inboundMsg),
+    )
+
+    ## Let the handler reach the decrypt await, then close mid-flight.
+    await sleepAsync(50.milliseconds)
+    (await manager.closeChannel(channelId)).expect("closeChannel")
+
+    decryptGate.complete()
+    await sleepAsync(100.milliseconds)
+    check not fired
+
+    ## Restore pass-through decryption for the remaining tests.
+    Decrypt
+      .replaceProvider(
+        DefaultBrokerContext,
+        proc(payload: seq[byte]): Future[Result[Decrypt, string]] {.async.} =
+          return ok(Decrypt(payload)),
+      )
+      .expect("replaceProvider restore")
+
+    (await waku.stop()).expect("stop")
+
 suite "Reliable Channel - send state machine":
   asyncTest "MessageSentEvent finalises the channelReqId as Sent":
     ## Drives the real send pipeline (`send` -> segmentation -> SDS ->