fix: subscribe to the channel's content topic on channel_create (#4081)

logos_delivery/api/messaging_client_api.nim

@@ -16,3 +16,10 @@ RequestBroker:
   proc MessagingSend(
     envelope: MessageEnvelope
   ): Future[Result[RequestId, string]] {.async.}
+
+# Semi detached MessagingClient subscription interface, same pattern as MessagingSend.
+RequestBroker(sync):
+  proc MessagingSubscribe(contentTopic: ContentTopic): Result[void, string]
+
+RequestBroker(sync):
+  proc MessagingUnsubscribe(contentTopic: ContentTopic): Result[void, string]

logos_delivery/channels/api/channel_lifecycle.nim

@@ -4,6 +4,7 @@ import std/tables
 import results, chronos, chronicles
 
 import logos_delivery/api/types
+import logos_delivery/api/messaging_client_api
 import logos_delivery/channels/reliable_channel_manager
 import logos_delivery/channels/reliable_channel
 import logos_delivery/waku/persistency/sds_persistency
@@ -35,9 +36,19 @@ proc createReliableChannel*(
 ): Result[ChannelId, string] =
   ## Encryption and egress providers must be installed (or `setNoopEncryption()`)
   ## before traffic flows on the channel.
+  ## Subscribes to `contentTopic`; without a `MessagingSubscribe` provider the
+  ## subscription is deferred to `ReliableChannelManager.start`.
   if self.channels.hasKey(channelId):
     return err("channel already exists: " & channelId)
 
+  # Subscribe before constructing so a failure leaks no listeners.
+  if MessagingSubscribe.isProvided(self.brokerCtx):
+    MessagingSubscribe.request(self.brokerCtx, contentTopic).isOkOr:
+      return err("failed to subscribe to content topic: " & error)
+  else:
+    debug "no MessagingSubscribe provider, deferring content topic subscription",
+      channelId = channelId, contentTopic = contentTopic
+
   let cc = self.conf
   let segConfig = SegmentationConfig(
     segmentSizeBytes: cc.segmentationSegmentSizeBytes.get(DefaultSegmentSizeBytes),
@@ -74,10 +85,24 @@ proc closeChannel*(
     self: ReliableChannelManager, channelId: ChannelId
 ): Future[Result[void, string]] {.async: (raises: []).} =
   ## Stops the channel's SDS loops and releases the channel. Persisted SDS
-  ## state survives, so re-creating the channel restores it.
+  ## state survives, so re-creating the channel restores it. Unsubscribes the
+  ## content topic unless another open channel still uses it.
   let chn = self.channels.getOrDefault(channelId)
   if chn.isNil():
     return err("unknown channel: " & channelId)
   self.channels.del(channelId)
   await chn.stop()
+
+  # After `stop` so in-flight sends cannot auto-resubscribe; best-effort.
+  let contentTopic = chn.getContentTopic()
+  var topicStillUsed = false
+  for other in self.channels.values:
+    if other.getContentTopic() == contentTopic:
+      topicStillUsed = true
+      break
+  if not topicStillUsed and MessagingUnsubscribe.isProvided(self.brokerCtx):
+    MessagingUnsubscribe.request(self.brokerCtx, contentTopic).isOkOr:
+      warn "failed to unsubscribe closed channel's content topic",
+        channelId = channelId, contentTopic = contentTopic, error = error
+
   return ok()

logos_delivery/channels/reliable_channel_manager.nim

@@ -15,6 +15,7 @@ import brokers/broker_context
 
 import logos_delivery/api/types
 import logos_delivery/api/reliable_channel_manager_api
+import logos_delivery/api/messaging_client_api
 import logos_delivery/api/conf/channels_conf
 
 import ./reliable_channel
@@ -55,6 +56,15 @@ proc start*(self: ReliableChannelManager): Result[void, string] =
   ## payloads. `setProvider` refuses to overwrite, so an application that
   ## installed its own encryption before start keeps it.
   setNoopEncryption()
+
+  # Subscribe channels created before the MessagingSubscribe provider existed.
+  if MessagingSubscribe.isProvided(self.brokerCtx):
+    for chn in self.channels.values:
+      MessagingSubscribe.request(self.brokerCtx, chn.getContentTopic()).isOkOr:
+        warn "failed to subscribe channel's content topic",
+          channelId = chn.getChannelId(),
+          contentTopic = chn.getContentTopic(),
+          error = error
   ok()
 
 proc stop*(self: ReliableChannelManager) {.async.} =

logos_delivery/messaging/messaging_client_lifecycle.nim

@@ -8,6 +8,7 @@ import
   logos_delivery/messaging/api/send,
   logos_delivery/api/messaging_client_api,
   logos_delivery/waku/waku,
+  logos_delivery/waku/api/subscriptions,
   logos_delivery/messaging/delivery_service/[recv_service, send_service]
 
 # Surfaces the messaging API interface (and its Message* events) to consumers.
@@ -25,6 +26,18 @@ proc start*(self: MessagingClient): Result[void, string] =
       return await self.send(envelope),
   )
 
+  ?MessagingSubscribe.setProvider(
+    self.brokerCtx,
+    proc(contentTopic: ContentTopic): Result[void, string] =
+      self.waku.subscribe(contentTopic),
+  )
+
+  ?MessagingUnsubscribe.setProvider(
+    self.brokerCtx,
+    proc(contentTopic: ContentTopic): Result[void, string] =
+      self.waku.unsubscribe(contentTopic),
+  )
+
   self.started = true
   ok()
 
@@ -32,6 +45,8 @@ proc stop*(self: MessagingClient) {.async.} =
   if not self.started:
     return
   MessagingSend.clearProvider(self.brokerCtx)
+  MessagingSubscribe.clearProvider(self.brokerCtx)
+  MessagingUnsubscribe.clearProvider(self.brokerCtx)
   await self.sendService.stopSendService()
   await self.recvService.stopRecvService()
   self.started = false

tests/channels/test_channel_lifecycle.nim

@@ -1,11 +1,12 @@
 {.used.}
 
-import chronos, testutils/unittests
+import results, chronos, testutils/unittests
 import brokers/broker_context
 
 import ../testlib/[common, testasync]
 
 import logos_delivery/api/conf/channels_conf
+import logos_delivery/api/messaging_client_api
 import logos_delivery/channels/reliable_channel_manager
 import logos_delivery/channels/api/channel_lifecycle
 import logos_delivery/channels/encryption/noop_encryption
@@ -37,3 +38,53 @@ suite "Reliable Channel - lifecycle":
       (await manager.closeChannel(channelId)).expect("closeChannel")
 
       check not manager.channelExists(channelId)
+
+  asyncTest "create subscribes the content topic, close unsubscribes it":
+    ## Close unsubscribes only when no other open channel shares the topic.
+    const
+      topic = ContentTopic("/reliable-channel/test/subscribe/proto")
+      chnA = ChannelId("subscribe-channel-a")
+      chnB = ChannelId("subscribe-channel-b")
+
+    var subscribed: seq[ContentTopic]
+    var unsubscribed: seq[ContentTopic]
+    var manager: ReliableChannelManager
+    lockNewGlobalBrokerContext:
+      let brokerCtx = globalBrokerContext()
+      manager = ReliableChannelManager.new(ReliableChannelManagerConf()).expect(
+          "ReliableChannelManager.new"
+        )
+      setNoopEncryption()
+
+      MessagingSubscribe
+        .setProvider(
+          brokerCtx,
+          proc(contentTopic: ContentTopic): Result[void, string] =
+            subscribed.add(contentTopic)
+            ok(),
+        )
+        .expect("setProvider MessagingSubscribe")
+      MessagingUnsubscribe
+        .setProvider(
+          brokerCtx,
+          proc(contentTopic: ContentTopic): Result[void, string] =
+            unsubscribed.add(contentTopic)
+            ok(),
+        )
+        .expect("setProvider MessagingUnsubscribe")
+
+      discard manager.createReliableChannel(chnA, topic, SdsParticipantID("a")).expect(
+          "createReliableChannel a"
+        )
+      check subscribed == @[topic]
+
+      discard manager.createReliableChannel(chnB, topic, SdsParticipantID("b")).expect(
+          "createReliableChannel b"
+        )
+
+      ## chnB still uses the topic: closing chnA must not unsubscribe.
+      (await manager.closeChannel(chnA)).expect("closeChannel a")
+      check unsubscribed.len == 0
+
+      (await manager.closeChannel(chnB)).expect("closeChannel b")
+      check unsubscribed == @[topic]

tests/channels/test_reliable_channel_send_receive.nim

@@ -1315,3 +1315,47 @@ suite "Reliable Channel - SDS protocol semantics":
     check (await manager.closeChannel(ChannelId("no-such-channel"))).isErr()
 
     (await waku.stop()).expect("stop")
+
+suite "Reliable Channel - content topic subscription":
+  asyncTest "channels subscribe on create/start and unsubscribe on close":
+    ## Without a subscription `RecvService` drops the channel's inbound traffic.
+    const
+      preStartChannelId = ChannelId("sub-pre-start-channel")
+      preStartTopic = ContentTopic("/reliable-channel/1/sub-pre-start/proto")
+      postStartChannelId = ChannelId("sub-post-start-channel")
+      postStartTopic = ContentTopic("/reliable-channel/1/sub-post-start/proto")
+
+    var waku: LogosDelivery
+    var manager: ReliableChannelManager
+    lockNewGlobalBrokerContext:
+      waku = (await LogosDelivery.new(createApiNodeConf())).expect("LogosDelivery.new")
+      manager = waku.reliableChannelManager
+
+      setNoopEncryption()
+
+      ## Created before start: subscribed by `ReliableChannelManager.start`.
+      discard manager
+        .createReliableChannel(
+          preStartChannelId, preStartTopic, SdsParticipantID("local")
+        )
+        .expect("createReliableChannel pre-start")
+      check not waku.waku.isSubscribed(preStartTopic).expect("isSubscribed")
+
+      (await waku.start()).expect("start")
+      check waku.waku.isSubscribed(preStartTopic).expect("isSubscribed")
+
+      ## Created after start: subscribed immediately.
+      discard manager
+        .createReliableChannel(
+          postStartChannelId, postStartTopic, SdsParticipantID("local")
+        )
+        .expect("createReliableChannel post-start")
+      check waku.waku.isSubscribed(postStartTopic).expect("isSubscribed")
+
+      (await manager.closeChannel(postStartChannelId)).expect("closeChannel post-start")
+      check not waku.waku.isSubscribed(postStartTopic).expect("isSubscribed")
+
+      (await manager.closeChannel(preStartChannelId)).expect("closeChannel pre-start")
+      check not waku.waku.isSubscribed(preStartTopic).expect("isSubscribed")
+
+    (await waku.stop()).expect("stop")