fix(dm): harden pager loop guard and foreground-only status writes

Audit follow-ups to the demand-driven paging change: - UntilLimitPager.onEose: require the reached cursor to move strictly older each page. A misbehaving relay that returns events but none older than already reached would otherwise pin the cursor and the on-screen sentinel would re-request the same window forever; treat it as the bottom. - updateStatus (gift-wrap + rooms-list NIP-04): only write the shared display StateFlows for the foreground account, mirroring the existing exhausted guard, so a background account's late EOSE can't clobber the on-screen relay count / reached-back / per-relay markers. - Add UntilLimitPagerTest covering the requested/reached split, park-on- EOSE, done-on-empty, the strict-older guard, and per-relay independence.
2026-08-10 00:16:59 +00:00 · 2026-06-04 20:30:19 +00:00
parent d20b02047b
commit eaddceba16
4 changed files with 129 additions and 1 deletions
@@ -147,7 +147,16 @@ class UntilLimitPager<K> {
        if (c.pageCount == 0) {
            c.done = true
        } else {
-            c.reachedUntil = c.pageOldest
+            // The reached cursor must move strictly older every page (the next page asks `until =
+            // reached - 1`). A relay that returns events but none older than we already have — a
+            // misbehaving relay echoing the same newest events — would otherwise pin the cursor and the
+            // on-screen sentinel would re-request the same window forever. Treat that as the bottom.
+            val prev = c.reachedUntil
+            if (prev == null || c.pageOldest < prev) {
+                c.reachedUntil = c.pageOldest
+            } else {
+                c.done = true
+            }
        }
    }

@@ -185,6 +185,9 @@ class AccountGiftWrapsHistoryEoseManager(
    }

    private fun updateStatus(user: User) {
+        // The display flows are singletons shown for the foreground account; a background account's late
+        // EOSE must not overwrite them (its cursors still advance in the pager).
+        if (activeUser != user.pubkeyHex) return
        val relays = accounts[user.pubkeyHex]?.dmRelays?.flow?.value ?: emptySet()
        _relayCount.value = loadTracker.count()
        val start = startUntil()
@@ -162,6 +162,9 @@ class ChatroomListNip04HistorySubAssembler(
    }

    private fun updateStatus(user: User) {
+        // The display flows are singletons shown for the foreground account; a background account's late
+        // EOSE must not overwrite them (its cursors still advance in the pager).
+        if (activeUser != user.pubkeyHex) return
        val account = accounts[user.pubkeyHex]
        val relays = account?.let { allRelays(it) } ?: emptySet()
        _relayCount.value = loadTracker.count()
@@ -0,0 +1,113 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.service.relayClient.eoseManagers
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class UntilLimitPagerTest {
+    private val key = "acct"
+    private val relayA = RelayUrlNormalizer.normalizeOrNull("wss://a.relay")!!
+    private val relayB = RelayUrlNormalizer.normalizeOrNull("wss://b.relay")!!
+    private val start = 1_000L
+
+    @Test
+    fun unarmedRelayIsNotRequestedButCountsAsActive() {
+        val pager = UntilLimitPager<String>()
+        assertFalse(pager.isArmed(key, relayA))
+        assertEquals(emptyList<Any>(), pager.armedRelays(key, listOf(relayA)))
+        // not done, so still "active" (there is history to ask for once advanced)
+        assertEquals(listOf(relayA), pager.activeRelays(key, listOf(relayA)))
+        // marker sits at the floor until it delivers
+        assertEquals(start, pager.reachedUntilFor(key, relayA, start))
+    }
+
+    @Test
+    fun firstAdvanceRequestsTheFloorThenSubsequentPagesStepBelowReached() {
+        val pager = UntilLimitPager<String>()
+
+        assertTrue(pager.advance(key, relayA, start))
+        assertTrue(pager.isArmed(key, relayA))
+        assertEquals(start, pager.requestedUntilFor(key, relayA))
+
+        // page returns events; oldest seen = 800
+        pager.onEvent(key, relayA, 900)
+        pager.onEvent(key, relayA, 800)
+        pager.onEose(key, relayA)
+        assertEquals(800L, pager.reachedUntilFor(key, relayA, start))
+        // EOSE does NOT move the requested cursor — the relay parks at the same filter
+        assertEquals(start, pager.requestedUntilFor(key, relayA))
+
+        // next advance steps to reached - 1
+        assertTrue(pager.advance(key, relayA, start))
+        assertEquals(799L, pager.requestedUntilFor(key, relayA))
+    }
+
+    @Test
+    fun emptyPageMarksRelayDoneAndBlocksFurtherAdvance() {
+        val pager = UntilLimitPager<String>()
+        pager.advance(key, relayA, start)
+        pager.onEose(key, relayA) // no events
+        assertTrue(pager.isDone(key, relayA))
+        assertFalse(pager.advance(key, relayA, start))
+        assertEquals(emptyList<Any>(), pager.activeRelays(key, listOf(relayA)))
+        assertEquals(emptyList<Any>(), pager.armedRelays(key, listOf(relayA)))
+    }
+
+    @Test
+    fun aPageThatDoesNotStepOlderEndsTheRelayInsteadOfLooping() {
+        val pager = UntilLimitPager<String>()
+        pager.advance(key, relayA, start)
+        pager.onEvent(key, relayA, 800)
+        pager.onEose(key, relayA)
+        assertEquals(800L, pager.reachedUntilFor(key, relayA, start))
+
+        // misbehaving relay: next page echoes an event no older than what we already reached
+        pager.advance(key, relayA, start) // requested = 799
+        pager.onEvent(key, relayA, 900) // newer than reached(800) — not strictly older
+        pager.onEose(key, relayA)
+        assertTrue("a non-advancing page should end the relay, not re-loop", pager.isDone(key, relayA))
+        assertEquals(800L, pager.reachedUntilFor(key, relayA, start))
+    }
+
+    @Test
+    fun relaysAreTrackedIndependently() {
+        val pager = UntilLimitPager<String>()
+        pager.advance(key, relayA, start)
+        pager.onEvent(key, relayA, 500)
+        pager.onEose(key, relayA)
+        // B never advanced
+        assertEquals(listOf(relayA), pager.armedRelays(key, listOf(relayA, relayB)))
+        assertEquals(500L, pager.reachedUntilFor(key, relayA, start))
+        assertEquals(start, pager.reachedUntilFor(key, relayB, start))
+        // deepest reached across both = A's 500 (B counts as the floor)
+        assertEquals(500L, pager.deepestReached(key, listOf(relayA, relayB), start))
+    }
+
+    @Test
+    fun deepestReachedIsNullWhenNoRelays() {
+        val pager = UntilLimitPager<String>()
+        assertEquals(null, pager.deepestReached(key, emptyList(), start))
+    }
+}