From dd259bab333c113458d12b48e5bcd069feca296a Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Thu, 8 Jan 2026 10:00:42 -0500 Subject: [PATCH] Separates currentFilters from lastSeenFilters to correctly close subscriptions with relays when ending them on NostrClient. --- .../relay/client/pool/PoolRequests.kt | 201 +++++++++++------- .../relay/client/reqs/IRequestListener.kt | 2 + .../client/reqs/RelayActiveRequestStates.kt | 2 +- .../client/reqs/RequestSubscriptionState.kt | 36 +++- .../subscriptions/SubscriptionController.kt | 2 +- 5 files changed, 153 insertions(+), 90 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt index 4847a747c..ac66e0731 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt @@ -36,21 +36,39 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.flow.MutableStateFlow +/** + * Manages relay subscriptions for the entire pool in a way that only + * sends new states to the relay if they have changed. + * + * This code also awaits a subscription to come to EOSE since many relays + * have through switching subs while they are processing the past. + */ class PoolRequests { /** - * Desired subs and listeners are changed immediately when the local code requests + * Desired subs and listeners + * + * These are changed immediately when the local code requests and should map + * to what the app whants to do. */ private val desiredSubs = LargeCache>>() private val desiredSubListeners = LargeCache() val desiredRelays = MutableStateFlow(setOf()) /** - * relay states are kept and only removed after everything is processed. + * Relay states map what we think the relay is doing + * + * This is important to link responses with which filter was a sub replying to and + * to figure out if we need to update the relay if a REQ has changed. */ private val relayState = LargeCache>() fun subState(subId: String): RequestSubscriptionState = relayState.getOrCreate(subId) { RequestSubscriptionState() } + /** + * This is called when a sub is added or removed from this class and + * should update the desired relay list to get the pool to connect + * to new ones or disconnect to old ones if they are not needed anymore + */ private fun updateRelays() { val myRelays = mutableSetOf() desiredSubs.forEach { sub, perRelayFilters -> @@ -62,19 +80,23 @@ class PoolRequests { } } + /** + * Returns all the active filters for a relay, per subscription + */ fun activeFiltersFor(url: NormalizedRelayUrl): Map> { val myRelays = mutableMapOf>() desiredSubs.forEach { sub, perRelayFilters -> val filters = perRelayFilters[url] if (filters != null) { - myRelays.put(sub, filters) + myRelays[sub] = filters } } return myRelays } /** - * Adds a new filter to the pool, and returns the relays that need to be updated. + * Adds a new subscription to the pool, and returns which relays + * MIGHT need to be updated. */ fun addOrUpdate( subId: String, @@ -98,7 +120,8 @@ class PoolRequests { } /** - * Removes the sub from the pool, and returns the relays that need to be updated. + * Removes the sub from the pool, and returns which relays + * MIGHT need to be updated. */ fun remove(subId: String): Set = if (desiredSubs.containsKey(subId)) { @@ -108,8 +131,6 @@ class PoolRequests { desiredSubs.remove(subId) // update listener desiredSubListeners.remove(subId) - // remove states - relayState.remove(subId) // update relays for pool updateRelays() // return all affected relays @@ -123,12 +144,102 @@ class PoolRequests { // -------------------------- // State management functions // -------------------------- + + /** + * When a connecting, updates the state of all subs + */ fun onConnecting(url: NormalizedRelayUrl) { + // Change states to connecting. relayState.forEach { subId, state -> state.connecting(url) } } + /** + * When a new command is sent to this relay, updates the state + */ + fun onSent( + relay: NormalizedRelayUrl, + cmd: Command, + ) { + when (cmd) { + is ReqCmd -> { + subState(cmd.subId).onOpenReq(relay, cmd.filters) + desiredSubListeners.get(cmd.subId)?.onStartReq( + relay = relay.url, + forFilters = cmd.filters, + ) + } + is CloseCmd -> { + subState(cmd.subId).onCloseReq(relay) + desiredSubListeners.get(cmd.subId)?.onCloseReq( + relay = relay.url, + ) + } + } + } + + /** + * When a new message is received by the relay, updates the sub + */ + fun onIncomingMessage( + relay: IRelayClient, + msg: Message, + ) { + when (msg) { + is EventMessage -> { + val state = relayState.get(msg.subId) + state?.onNewEvent(relay.url) + desiredSubListeners.get(msg.subId)?.onEvent( + event = msg.event, + isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE, + relay = relay.url, + forFilters = state?.lastKnownFilterStates(relay.url), + ) + } + is EoseMessage -> { + val state = relayState.get(msg.subId) + state?.onEose(relay.url) + desiredSubListeners.get(msg.subId)?.onEose( + relay = relay.url, + forFilters = state?.lastKnownFilterStates(relay.url), + ) + + // send a newer version when done + sendToRelayIfChanged(msg.subId, relay.url) { cmd -> + relay.sendOrConnectAndSync(cmd) + } + } + is ClosedMessage -> { + val state = relayState.get(msg.subId) + state?.onClosed(relay.url) + + desiredSubListeners.get(msg.subId)?.onClosed( + message = msg.message, + relay = relay.url, + forFilters = state?.lastKnownFilterStates(relay.url), + ) + + // send a newer version when done + sendToRelayIfChanged(msg.subId, relay.url) { cmd -> + // don't send a close if just closed + if (cmd !is CloseCmd) { + relay.sendOrConnectAndSync(cmd) + } + } + } + } + } + + /** + * When the relay disconnects + */ + fun onDisconnected(url: NormalizedRelayUrl) { + relayState.forEach { subId, state -> + state.disconnected(url) + } + } + fun syncState( relay: NormalizedRelayUrl, sync: (Command) -> Unit, @@ -143,77 +254,6 @@ class PoolRequests { } } - fun onSent( - relay: NormalizedRelayUrl, - cmd: Command, - ) { - when (cmd) { - is ReqCmd -> { - subState(cmd.subId).onOpenReq(relay, cmd.filters) - desiredSubListeners.get(cmd.subId)?.onStartReq( - relay = relay.url, - forFilters = cmd.filters, - ) - } - is CloseCmd -> subState(cmd.subId).onCloseReq(relay) - } - } - - fun onIncomingMessage( - relay: IRelayClient, - msg: Message, - ) { - when (msg) { - is EventMessage -> { - val state = relayState.get(msg.subId) - state?.onNewEvent(relay.url) - desiredSubListeners.get(msg.subId)?.onEvent( - event = msg.event, - isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE, - relay = relay.url, - forFilters = state?.currentFilters(relay.url), - ) - } - is EoseMessage -> { - val state = relayState.get(msg.subId) - state?.onEose(relay.url) - desiredSubListeners.get(msg.subId)?.onEose( - relay = relay.url, - forFilters = state?.currentFilters(relay.url), - ) - - // send a newer version when done - sendToRelayIfChanged(msg.subId, relay.url) { cmd -> - relay.sendOrConnectAndSync(cmd) - } - } - is ClosedMessage -> { - val state = relayState.get(msg.subId) - state?.onClosed(relay.url) - - desiredSubListeners.get(msg.subId)?.onClosed( - message = msg.message, - relay = relay.url, - forFilters = state?.currentFilters(relay.url), - ) - - // send a newer version when done - sendToRelayIfChanged(msg.subId, relay.url) { cmd -> - // don't send a close if just closed - if (cmd !is CloseCmd) { - relay.sendOrConnectAndSync(cmd) - } - } - } - } - } - - fun onDisconnected(url: NormalizedRelayUrl) { - relayState.forEach { subId, state -> - state.disconnected(url) - } - } - /** * If cannot connect, closes subs */ @@ -225,7 +265,7 @@ class PoolRequests { desiredSubListeners.get(subId)?.onCannotConnect( message = errorMessage, relay = url, - forFilters = state.currentFilters(url), + forFilters = state.lastKnownFilterStates(url), ) } } @@ -258,7 +298,8 @@ class PoolRequests { relay: NormalizedRelayUrl, sync: (Command) -> Unit, ) { - val oldFilters = relayState.get(subId)?.currentFilters(relay) + val state = relayState.get(subId) + val oldFilters = state?.currentFilters(relay) val newFilters = desiredSubs.get(subId)?.get(relay) sendToRelayIfChanged(subId, oldFilters, newFilters, sync) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/IRequestListener.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/IRequestListener.kt index 57f114c3e..64d321fc3 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/IRequestListener.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/IRequestListener.kt @@ -53,4 +53,6 @@ interface IRequestListener { relay: String, forFilters: List, ) {} + + fun onCloseReq(relay: String) {} } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RelayActiveRequestStates.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RelayActiveRequestStates.kt index b27dc1efd..7ddb3a2db 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RelayActiveRequestStates.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RelayActiveRequestStates.kt @@ -43,7 +43,7 @@ class RelayActiveRequestStates( private val clientListener = object : IRelayClientListener { override fun onConnecting(relay: IRelayClient) { - subStates.put(relay.url, RequestSubscriptionState()) + subStates[relay.url] = RequestSubscriptionState() } override fun onIncomingMessage( diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt index f2c7b253b..390265219 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/reqs/RequestSubscriptionState.kt @@ -22,6 +22,10 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.reqs import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +/** + * Manages the State of Subscriptions by logging states as the + * subscription progresses. + */ class RequestSubscriptionState { // Logs the state of each channel to: // 1. inform when an event is received as live @@ -32,36 +36,53 @@ class RequestSubscriptionState { private val subStates = mutableMapOf() private val filterStates = mutableMapOf>() + /** + * This cache is used to make sure we know what the relay was processing + * before a close or disconnect so that if new events still arrive + * we can link them with the appropriate filters. + */ + private val lastKnownFilterStates = mutableMapOf>() + fun currentFilters() = filterStates fun currentFilters(reference: T) = filterStates[reference] + fun lastKnownFilterStates(reference: T) = lastKnownFilterStates[reference] + fun currentState(reference: T) = subStates[reference] fun onNewEvent(reference: T) { if (subStates[reference] == ReqSubStatus.SENT) { - subStates.put(reference, ReqSubStatus.QUERYING_PAST) + subStates[reference] = ReqSubStatus.QUERYING_PAST } } fun onEose(reference: T) { - subStates.put(reference, ReqSubStatus.LIVE) + subStates[reference] = ReqSubStatus.LIVE } fun onClosed(reference: T) { - subStates.put(reference, ReqSubStatus.CLOSED) + subStates[reference] = ReqSubStatus.CLOSED + // Closed messages are usually relays refusing to process a REQ + // This message keeps the state of filterStates intact to + // avoid sending the same filter, and getting immediately closed, + // over and over again. + + // filterStates.remove(reference) } fun onOpenReq( reference: T, filters: List, ) { - subStates.put(reference, ReqSubStatus.SENT) - filterStates.put(reference, filters) + subStates[reference] = ReqSubStatus.SENT + filterStates[reference] = filters + lastKnownFilterStates[reference] = filters } fun onCloseReq(reference: T) { - subStates.put(reference, ReqSubStatus.CLOSED) + subStates[reference] = ReqSubStatus.CLOSED + filterStates.remove(reference) } fun connecting(reference: T) { @@ -70,8 +91,7 @@ class RequestSubscriptionState { } fun disconnected(reference: T) { - // Can't remove filterStates because disconnections happen - // before processing all the events in the channel queue. subStates.remove(reference) + filterStates.remove(reference) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/subscriptions/SubscriptionController.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/subscriptions/SubscriptionController.kt index c87bb5158..58930f47f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/subscriptions/SubscriptionController.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/subscriptions/SubscriptionController.kt @@ -59,9 +59,9 @@ class SubscriptionController( fun dismissSubscription(subId: String) = getSub(subId)?.let { dismissSubscription(it) } fun dismissSubscription(subscription: Subscription) { - client.close(subscription.id) subscription.reset() subscriptions.remove(subscription.id) + client.close(subscription.id) } fun updateRelays() {