diff --git a/logos_delivery/api/messaging_client_api.nim b/logos_delivery/api/messaging_client_api.nim index 5906ab564..03f88175a 100644 --- a/logos_delivery/api/messaging_client_api.nim +++ b/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] diff --git a/logos_delivery/channels/api/channel_lifecycle.nim b/logos_delivery/channels/api/channel_lifecycle.nim index 57d4e83fe..6b233faca 100644 --- a/logos_delivery/channels/api/channel_lifecycle.nim +++ b/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() diff --git a/logos_delivery/channels/reliable_channel_manager.nim b/logos_delivery/channels/reliable_channel_manager.nim index 13f5d5937..16c4ddbff 100644 --- a/logos_delivery/channels/reliable_channel_manager.nim +++ b/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.} = diff --git a/logos_delivery/messaging/messaging_client_lifecycle.nim b/logos_delivery/messaging/messaging_client_lifecycle.nim index 52ea295aa..9fd275b92 100644 --- a/logos_delivery/messaging/messaging_client_lifecycle.nim +++ b/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 diff --git a/tests/channels/test_channel_lifecycle.nim b/tests/channels/test_channel_lifecycle.nim index c5cf810cf..33f94f7e0 100644 --- a/tests/channels/test_channel_lifecycle.nim +++ b/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] diff --git a/tests/channels/test_reliable_channel_send_receive.nim b/tests/channels/test_reliable_channel_send_receive.nim index d72fb4d43..41ce46327 100644 --- a/tests/channels/test_reliable_channel_send_receive.nim +++ b/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")