Separates currentFilters from lastSeenFilters to correctly close subscriptions with relays when ending them on NostrClient.

2026-01-08 10:00:42 -05:00
parent 19d3c5a10e
commit dd259bab33
5 changed files with 153 additions and 90 deletions
@@ -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<String, Map<NormalizedRelayUrl, List<Filter>>>()
    private val desiredSubListeners = LargeCache<String, IRequestListener>()
    val desiredRelays = MutableStateFlow(setOf<NormalizedRelayUrl>())

    /**
-     * 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<String, RequestSubscriptionState<NormalizedRelayUrl>>()

    fun subState(subId: String): RequestSubscriptionState<NormalizedRelayUrl> = 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<NormalizedRelayUrl>()
        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<String, List<Filter>> {
        val myRelays = mutableMapOf<String, List<Filter>>()
        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<NormalizedRelayUrl> =
        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)
    }
@@ -53,4 +53,6 @@ interface IRequestListener {
        relay: String,
        forFilters: List<Filter>,
    ) {}
+
+    fun onCloseReq(relay: String) {}
 }
@@ -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(
@@ -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<T> {
    // Logs the state of each channel to:
    // 1. inform when an event is received as live
@@ -32,36 +36,53 @@ class RequestSubscriptionState<T> {
    private val subStates = mutableMapOf<T, ReqSubStatus>()
    private val filterStates = mutableMapOf<T, List<Filter>>()

+    /**
+     * 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<T, List<Filter>>()
+
    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<Filter>,
    ) {
-        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<T> {
    }

    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)
    }
 }
@@ -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() {