From cf350569d417a9311c037932e0d2020b8d8d31d5 Mon Sep 17 00:00:00 2001 From: Ocnrb Date: Sun, 27 Sep 2026 17:44:10 +0100 Subject: [PATCH 1/2] Treat edits and deletions that did not come back as owed, not as none - P1 gets the content's ceiling, and a P1 cut by its budget is read again - An older page is taken only when P0 and P1 both answered, with a backoff before asking again - The open's read is marked as still out, so nothing calls the channel empty meanwhile Co-authored-by: Claude Opus 5.5 --- app/src/main/assets/pombo_bridge.html | 5 +- .../java/com/pombo/android/AppViewModel.kt | 2 + .../java/com/pombo/android/ChannelManager.kt | 95 ++++++- .../com/pombo/android/OverridesOwedTest.kt | 248 ++++++++++++++++++ 4 files changed, 338 insertions(+), 12 deletions(-) create mode 100644 app/src/test/java/com/pombo/android/OverridesOwedTest.kt diff --git a/app/src/main/assets/pombo_bridge.html b/app/src/main/assets/pombo_bridge.html index ccf0196..d7d0c7c 100644 --- a/app/src/main/assets/pombo_bridge.html +++ b/app/src/main/assets/pombo_bridge.html @@ -1560,6 +1560,7 @@ ); const all = []; const opener = api._sealedOpener(pwKey); + const drain = { streamId: a.streamId, partition: a.partition }; const partial = await api._drainResend(stream, function (msg) { if (!a.recoverSigner && !api._envelopeIsAuthentic(msg)) return; const rec = { @@ -1573,7 +1574,7 @@ storageFetch.stamp(rec.meta, msg, a.streamId, a.partition); all.push(rec); opener.take(rec); - }, Date.now() + (a.budgetMs || 40000), { streamId: a.streamId, partition: a.partition }); + }, Date.now() + (a.budgetMs || 40000), drain); await opener.done(); // Exhaustion is about what the RANGE held, not about what survived the // permission filter: a page made only of refused rows still means there @@ -1587,7 +1588,7 @@ // A cut-short drain cannot prove exhaustion — keep pagination alive. return { messages: all.slice(start), hasMore: fetched > count || partial, partial: partial, - readError: storageFetch.lastReadError(a.streamId, a.partition) + errors: drain.errors, readError: storageFetch.lastReadError(a.streamId, a.partition) }; }, diff --git a/app/src/main/java/com/pombo/android/AppViewModel.kt b/app/src/main/java/com/pombo/android/AppViewModel.kt index 78b9384..0f64cad 100644 --- a/app/src/main/java/com/pombo/android/AppViewModel.kt +++ b/app/src/main/java/com/pombo/android/AppViewModel.kt @@ -1216,6 +1216,8 @@ class AppViewModel(app: Application) : AndroidViewModel(app), PomboBridge.Listen val hasMoreHistory get() = manager.hasMoreHistory val historyError get() = manager.historyError val historyRead get() = manager.historyRead + val overridesOwed get() = manager.overridesOwed + val pagingNudge get() = manager.pagingNudge val waitingForKeys get() = manager.waitingForKeys val loadingHistory get() = manager.loadingHistory val isPreview get() = manager.isPreview diff --git a/app/src/main/java/com/pombo/android/ChannelManager.kt b/app/src/main/java/com/pombo/android/ChannelManager.kt index f574f57..4027de1 100644 --- a/app/src/main/java/com/pombo/android/ChannelManager.kt +++ b/app/src/main/java/com/pombo/android/ChannelManager.kt @@ -421,7 +421,10 @@ class ChannelManager( val channel = _current.value?.takeIf { it.messageStreamId == messageStreamId } ?: return@withContext Log.i(TAG, "refreshing history") - loadHistory(channel, switchGeneration) + val generation = switchGeneration + if (!loadHistory(channel, generation) && _historyRead.value == HistoryRead.OK) { + retryOpenReads(channel, generation) + } // The -3 artifacts fetched at open were epoch-sealed and // unreadable until this key arrived — pins/moderation and a // hidden channel's image need their own re-pull (the admin @@ -4873,7 +4876,19 @@ class ChannelManager( } val loads = launch { listOf( - launch { if (!loadHistory(channel, generation)) retryOpenReads(channel, generation) }, + launch { + // The gate below lets go at its safety net; the + // empty state waits for this read instead. + if (stillCurrent(generation)) _historyRead.value = HistoryRead.READING + val readAll = try { + loadHistory(channel, generation) + } finally { + if (stillCurrent(generation) && _historyRead.value == HistoryRead.READING) { + _historyRead.value = HistoryRead.OK + } + } + if (!readAll) retryOpenReads(channel, generation) + }, launch { loadAdminState(channel, generation) } ).joinAll() } @@ -5002,6 +5017,8 @@ class ChannelManager( _loadingHistory.value = false _historyError.value = null _historyRead.value = HistoryRead.OK + _overridesOwed.value = false + resetPagingBackoff() // Both are per-channel verdicts — carrying them across a switch // shows the previous room's key/subscription state on this one. _waitingForKeys.value = false @@ -5101,6 +5118,7 @@ class ChannelManager( adminPollJob?.cancel(); adminPollJob = null memberCatchUpJob?.cancel(); memberCatchUpJob = null historyRetryJob?.cancel(); historyRetryJob = null + resetPagingBackoff() _current.value = null _isPreview.value = false unsubscribeAll(channel) @@ -5428,19 +5446,26 @@ class ChannelManager( private class HistoryPage( val entries: JSONArray, val contents: List, val readError: HistoryError? = null, /** The node broke the read off without answering it: what came is applied, the read is owed. */ - val failed: Boolean = false + val failed: Boolean = false, + /** The drain stopped before the page ended, by its budget or a break, with no refusal to explain it. */ + val cut: Boolean = false ) private val HistoryPage?.answered get() = this != null && !failed + /** Overrides owe every row of their page: one cut by its budget leaves edits and deletions unread. */ + private val HistoryPage?.whole get() = answered && this?.cut != true private val _historyError = MutableStateFlow(null) /** The refusal behind an empty history, for the empty state to explain. */ val historyError: StateFlow = _historyError.asStateFlow() - /** Whether the open channel's reads came back, are being read again, or gave up. */ - enum class HistoryRead { OK, RETRYING, FAILED } + /** Whether the open channel's reads came back, are still out, are being read again, or gave up. */ + enum class HistoryRead { OK, READING, RETRYING, FAILED } private val _historyRead = MutableStateFlow(HistoryRead.OK) val historyRead: StateFlow = _historyRead.asStateFlow() + private val _overridesOwed = MutableStateFlow(false) + /** The open channel's edits and deletions did not come back with its messages. */ + val overridesOwed: StateFlow = _overridesOwed.asStateFlow() private var historyRetryJob: Job? = null @Volatile private var historyRetryWake: CompletableDeferred? = null internal var historyRetryDelaysMs: LongArray = HISTORY_RETRY_DELAYS_MS @@ -5545,8 +5570,9 @@ class ChannelManager( } // A 4xx is the node's answer about this read; any other break // (network, 5xx, a page refused for its storedAt) left it unread. - val failed = res.optInt("errors", 0) > 0 && (readError == null || readError.status !in 400..499) - HistoryPage(arr, contents, readError, failed) + val refused = readError != null && readError.status in 400..499 + val failed = res.optInt("errors", 0) > 0 && !refused + HistoryPage(arr, contents, readError, failed, cut = res.optBoolean("partial", false) && !refused) } } catch (e: Exception) { Log.w(TAG, "history ${channel.name} P$partition: read failed: ${e.message}") @@ -5593,8 +5619,10 @@ class ChannelManager( onNoStorage = { noStorage += "content" } ) } + // The content's ceiling: nothing is applied before both pages are + // back, so this only lengthens an open whose node is slow on P1 alone. val o = async { - fetchHistoryPage(channel, StreamConstants.P_CONTROL, 50, 20_000, 30_000, + fetchHistoryPage(channel, StreamConstants.P_CONTROL, 50, 45_000, 60_000, onNoStorage = { noStorage += "overrides" }) } // Reactions moved to the -5 — without this read a reopened channel @@ -5693,8 +5721,9 @@ class ChannelManager( // without a `messages` array, which skipped this sweep and left any // override that had arrived before its target parked forever. applyPendingOverrides() - return (content.answered || "content" in noStorage) && - (overrides.answered || "overrides" in noStorage) && sideReadsOk + val overridesRead = overrides.whole || "overrides" in noStorage + _overridesOwed.value = !overridesRead + return (content.answered || "content" in noStorage) && overridesRead && sideReadsOk } /** Moves the pagination cursor back. Chunks count too — they carry no id. */ @@ -5703,6 +5732,42 @@ class ChannelManager( if (ts > 0L && (oldestTimestamp == 0L || ts < oldestTimestamp)) oldestTimestamp = ts } + private var pagingFailures = 0 + private var pagingRetryAt = 0L + private var pagingNudgeJob: Job? = null + internal var pagingRetryDelaysMs: LongArray = HISTORY_RETRY_DELAYS_MS + private val _pagingNudge = MutableStateFlow(0) + /** Ticks when an older page that came back incomplete may be asked for again. */ + val pagingNudge: StateFlow = _pagingNudge.asStateFlow() + + private fun resetPagingBackoff() { + pagingNudgeJob?.cancel(); pagingNudgeJob = null + pagingFailures = 0 + pagingRetryAt = 0L + } + + /** Asks for the page again after a backoff; past the last one, only the user's scroll does. */ + private fun backOffPaging(generation: Int) { + val wait = pagingRetryDelaysMs[minOf(pagingFailures, pagingRetryDelaysMs.size - 1)] + pagingFailures++ + pagingRetryAt = System.currentTimeMillis() + wait + pagingNudgeJob?.cancel(); pagingNudgeJob = null + if (pagingFailures > pagingRetryDelaysMs.size) return + pagingNudgeJob = scope.launch { + delay(wait) + if (stillCurrent(generation)) _pagingNudge.value++ + } + } + + /** Whether a range page is the node's answer: a refusal is; a break is not, nor a cut where every row counts. */ + private fun rangeAnswered(page: JSONObject?, cutCounts: Boolean): Boolean { + page ?: return false + val status = readErrorOf(page)?.status + if (status != null && status in 400..499) return true + if (page.optInt("errors", 0) > 0) return false + return !cutCounts || !page.optBoolean("partial", false) + } + /** * Loads the previous page of history (web: channels.js loadMoreHistory). * Fetches content (P0) and overrides (P1) older than the current cursor; @@ -5716,6 +5781,7 @@ class ChannelManager( // A page that failed keeps paging open, so an empty screen would ask // again every frame; the open's reads are the retry's until they are back. if (_historyRead.value != HistoryRead.OK) return 0 + if (System.currentTimeMillis() < pagingRetryAt) return 0 // Cursor of Date.now() when history held only reactions/chunks (web parity). val before = if (oldestTimestamp > 0L) oldestTimestamp else System.currentTimeMillis() @@ -5757,6 +5823,15 @@ class ChannelManager( // Discard if the user switched channels while we were fetching. if (!stillCurrent(generationAtStart)) return 0 + // Taken whole or not at all: content without its overrides paints + // edits and deletions undone, and the cursor would move past + // overrides that nothing reads again. + if (!rangeAnswered(content, cutCounts = false) || !rangeAnswered(overrides, cutCounts = true)) { + Log.w(TAG, "loadMore ${channel.name}: page came back incomplete, asking again later") + backOffPaging(generationAtStart) + return 0 + } + resetPagingBackoff() // Content first, then overrides — an edit/delete needs its target // present, so each partition gets its own batch rather than one diff --git a/app/src/test/java/com/pombo/android/OverridesOwedTest.kt b/app/src/test/java/com/pombo/android/OverridesOwedTest.kt new file mode 100644 index 0000000..746d20a --- /dev/null +++ b/app/src/test/java/com/pombo/android/OverridesOwedTest.kt @@ -0,0 +1,248 @@ +package com.pombo.android + +import com.pombo.android.core.StreamConstants +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.runBlocking +import org.json.JSONArray +import org.json.JSONObject +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +/** + * Edits and deletions live on P1, the messages on P0. A P1 that did not come + * back whole is owed, not empty: the open reads it again and says so + * meanwhile, and an older page is only taken with both halves. + */ +class OverridesOwedTest { + + private val owner = "0x" + "cd".repeat(20) + private val room = ChannelManagerHarness.channel("$owner/room-1") + private val h = ChannelManagerHarness(channels = listOf(room)) + private val manager = h.manager + private val t0 = System.currentTimeMillis() + + /** How a read answers: whole, not at all, cut by its budget, broken off, or refused by the node. */ + private enum class Answer { WHOLE, THROWS, CUT, BROKEN, REFUSED } + + @Volatile private var openP1 = Answer.WHOLE + private var content = listOf() + private var overrides = listOf() + /** Holds the open's P0 read until completed, when set. */ + @Volatile private var p0Gate: CompletableDeferred? = null + + @Volatile private var olderP0 = Answer.WHOLE + @Volatile private var olderP1 = Answer.WHOLE + private var olderContent = listOf() + private var olderOverrides = listOf() + + @Before fun setUp() { + manager.adminConfirmSleep = { } + manager.historyRetryDelaysMs = longArrayOf(600_000L, 600_000L, 600_000L, 600_000L) + manager.pagingRetryDelaysMs = longArrayOf(600_000L, 600_000L, 600_000L, 600_000L) + every { manager.adminFloorStore.get(any()) } returns null + coEvery { h.bridge.call("resend", any(), any()) } coAnswers { + val args = secondArg() + if (args.optString("streamId") != room.messageStreamId) { + page(emptyList()) + } else when (args.optInt("partition")) { + StreamConstants.P_MESSAGES -> { p0Gate?.await(); page(content) } + StreamConstants.P_CONTROL -> answer(overrides, openP1) + else -> page(emptyList()) + } + } + coEvery { h.bridge.call("resendRange", any(), any()) } answers { + val args = secondArg() + if (args.optString("streamId") != room.messageStreamId) { + page(emptyList()).put("hasMore", false) + } else when (args.optInt("partition")) { + StreamConstants.P_MESSAGES -> answer(olderContent, olderP0).put("hasMore", true) + StreamConstants.P_CONTROL -> answer(olderOverrides, olderP1).put("hasMore", false) + else -> page(emptyList()).put("hasMore", false) + } + } + } + + @After fun tearDown() = h.stop() + + private fun page(rows: List) = JSONObject().put("messages", JSONArray(rows.map { row -> + JSONObject().put("content", row) + .put("meta", JSONObject().put("publisherId", owner).put("timestamp", row.optLong("timestamp", t0))) + })) + + private fun answer(rows: List, how: Answer): JSONObject = when (how) { + Answer.WHOLE -> page(rows) + Answer.THROWS -> throw IllegalStateException("Timed out waiting for 60000 ms") + Answer.CUT -> page(emptyList()).put("partial", true) + Answer.BROKEN -> page(emptyList()).put("partial", true).put("errors", 1) + Answer.REFUSED -> page(emptyList()).put("partial", true).put("errors", 1) + .put("readError", JSONObject().put("status", 403).put("signed", true)) + } + + private fun text(id: String, at: Long) = JSONObject().put("type", "text").put("id", id) + .put("text", "hello $id").put("sender", owner).put("timestamp", at) + + private fun delete(targetId: String, at: Long) = JSONObject().put("type", "delete") + .put("targetId", targetId).put("timestamp", at) + + private fun ids() = manager.messages.value.map { it.id }.toSet() + + @Test + fun `a P1 that timed out is owed, and the read that brings it deletes what it should`() { + content = listOf(text("m1", t0 - 3), text("m2", t0 - 2)) + overrides = listOf(delete("m2", t0 - 1)) + openP1 = Answer.THROWS + + manager.openChannel(room.messageStreamId) + + assertEquals(ChannelManager.HistoryRead.RETRYING, manager.historyRead.value) + assertTrue(manager.overridesOwed.value) + assertEquals(setOf("m1", "m2"), ids()) + + openP1 = Answer.WHOLE + manager.kickOpenReadRetry() + + assertEquals(ChannelManager.HistoryRead.OK, manager.historyRead.value) + assertFalse(manager.overridesOwed.value) + assertEquals(setOf("m1"), ids()) + } + + @Test + fun `a P1 cut by its budget is owed, not an answer`() { + content = listOf(text("m1", t0 - 1)) + openP1 = Answer.CUT + + manager.openChannel(room.messageStreamId) + + assertEquals(ChannelManager.HistoryRead.RETRYING, manager.historyRead.value) + assertTrue(manager.overridesOwed.value) + } + + @Test + fun `a P1 the node refused is its answer`() { + content = listOf(text("m1", t0 - 1)) + openP1 = Answer.REFUSED + + manager.openChannel(room.messageStreamId) + + assertEquals(ChannelManager.HistoryRead.OK, manager.historyRead.value) + assertFalse(manager.overridesOwed.value) + } + + @Test + fun `giving up leaves the edits and deletions owed`() { + content = listOf(text("m1", t0 - 1)) + openP1 = Answer.THROWS + manager.openChannel(room.messageStreamId) + + repeat(manager.historyRetryDelaysMs.size) { manager.kickOpenReadRetry() } + + assertEquals(ChannelManager.HistoryRead.FAILED, manager.historyRead.value) + assertTrue(manager.overridesOwed.value) + } + + @Test + fun `while the open's read is out the channel is reading, and paging waits`() { + content = listOf(text("m1", t0 - 1)) + val gate = CompletableDeferred().also { p0Gate = it } + + manager.openChannel(room.messageStreamId) + + assertEquals(ChannelManager.HistoryRead.READING, manager.historyRead.value) + assertEquals(0, runBlocking { manager.loadMoreHistory() }) + coVerify(exactly = 0) { h.bridge.call("resendRange", any(), any()) } + + gate.complete(Unit) + + assertEquals(ChannelManager.HistoryRead.OK, manager.historyRead.value) + assertEquals(setOf("m1"), ids()) + } + + /** Opens with m3 and m4, both halves back, so the cursor sits before them. */ + private fun openWithRecentPage() { + content = listOf(text("m3", t0 - 3), text("m4", t0 - 2)) + olderContent = listOf(text("m1", t0 - 20), text("m2", t0 - 10)) + manager.openChannel(room.messageStreamId) + assertEquals(ChannelManager.HistoryRead.OK, manager.historyRead.value) + assertTrue(manager.hasMoreHistory.value) + } + + /** The page is left where it was, and the next ask within the backoff reads nothing. */ + private fun assertOlderPageNotTaken() { + assertEquals(0, runBlocking { manager.loadMoreHistory() }) + assertEquals(setOf("m3", "m4"), ids()) + assertTrue(manager.hasMoreHistory.value) + + assertEquals(0, runBlocking { manager.loadMoreHistory() }) + coVerify(exactly = 1) { + h.bridge.call("resendRange", match { + it.optString("streamId") == room.messageStreamId && it.optInt("partition") == StreamConstants.P_CONTROL + }, any()) + } + } + + @Test + fun `an older page is taken with its overrides`() { + openWithRecentPage() + olderOverrides = listOf(delete("m1", t0 - 15)) + + runBlocking { manager.loadMoreHistory() } + + assertEquals(setOf("m2", "m3", "m4"), ids()) + } + + @Test + fun `an older page whose P1 timed out is not taken`() { + openWithRecentPage() + olderP1 = Answer.THROWS + assertOlderPageNotTaken() + } + + @Test + fun `an older page whose P1 was cut by its budget is not taken`() { + openWithRecentPage() + olderP1 = Answer.CUT + assertOlderPageNotTaken() + } + + @Test + fun `an older page whose P1 broke off is not taken`() { + openWithRecentPage() + olderP1 = Answer.BROKEN + assertOlderPageNotTaken() + } + + @Test + fun `an older page whose P0 broke off is not taken`() { + openWithRecentPage() + olderP0 = Answer.BROKEN + assertOlderPageNotTaken() + } + + @Test + fun `an older P1 the node refused is its answer, and the page is taken`() { + openWithRecentPage() + olderP1 = Answer.REFUSED + + runBlocking { manager.loadMoreHistory() } + + assertEquals(setOf("m1", "m2", "m3", "m4"), ids()) + } + + @Test + fun `an incomplete older page is asked for again after each backoff, then left to the user`() { + manager.pagingRetryDelaysMs = longArrayOf(0L, 0L) + openWithRecentPage() + olderP1 = Answer.THROWS + + repeat(3) { runBlocking { manager.loadMoreHistory() } } + + assertEquals(2, manager.pagingNudge.value) + } +} From 77c2e20662ac239fca8e5a38c44a7681e18be6a2 Mon Sep 17 00:00:00 2001 From: Ocnrb Date: Sun, 27 Sep 2026 17:44:10 +0100 Subject: [PATCH 2/2] Say when edits and deletions are still loading or could not be loaded Co-authored-by: Claude Opus 5.5 --- .../pombo/android/ui/screens/ChatScreen.kt | 25 +++++++++++++++---- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/app/src/main/java/com/pombo/android/ui/screens/ChatScreen.kt b/app/src/main/java/com/pombo/android/ui/screens/ChatScreen.kt index 80d862a..c4e95e1 100644 --- a/app/src/main/java/com/pombo/android/ui/screens/ChatScreen.kt +++ b/app/src/main/java/com/pombo/android/ui/screens/ChatScreen.kt @@ -167,6 +167,8 @@ fun ChatScreen(vm: AppViewModel) { val hasMoreHistory by vm.hasMoreHistory.collectAsState() val historyError by vm.historyError.collectAsState() val historyRead by vm.historyRead.collectAsState() + val overridesOwed by vm.overridesOwed.collectAsState() + val pagingNudge by vm.pagingNudge.collectAsState() val loadingHistory by vm.loadingHistory.collectAsState() val loadingInitial by vm.initialLoad.collectAsState() val restoredTimeline by vm.restoredTimeline.collectAsState() @@ -232,7 +234,7 @@ fun ChatScreen(vm: AppViewModel) { val contentShort by remember { derivedStateOf { !listState.canScrollForward && !listState.canScrollBackward } } - LaunchedEffect(nearTop, contentShort, hasMoreHistory, loadingHistory, loadingInitial, messages.isEmpty(), searchOlder) { + LaunchedEffect(nearTop, contentShort, hasMoreHistory, loadingHistory, loadingInitial, messages.isEmpty(), searchOlder, pagingNudge) { // The reversed list is born at index 0 = the NEWEST message, with the // sentinel a full page away — `nearTop` cannot fire during opening, so // the chain-load that once parked whole channels at their oldest @@ -684,6 +686,18 @@ fun ChatScreen(vm: AppViewModel) { modifier = Modifier.fillMaxWidth().padding(horizontal = 16.dp, vertical = 8.dp) ) } + if (overridesOwed && visible.isNotEmpty()) { + Text( + if (historyRead == ChannelManager.HistoryRead.FAILED) { + "Edits and deletions could not be loaded. Reopen the channel" + } else { + "Loading edits and deletions…" + }, + color = Color.White.copy(alpha = 0.40f), fontSize = 13.sp, + textAlign = androidx.compose.ui.text.style.TextAlign.Center, + modifier = Modifier.fillMaxWidth().padding(horizontal = 16.dp, vertical = 8.dp) + ) + } Box(Modifier.weight(1f).fillMaxWidth()) { // Web renderMessages: while any loading signal is still live the @@ -691,7 +705,8 @@ fun ChatScreen(vm: AppViewModel) { // and still empty does it say "No messages yet". if (visible.isEmpty()) { val terminalEmpty = !loadingInitial && !loadingHistory && !hasMoreHistory && - historyRead != ChannelManager.HistoryRead.RETRYING + historyRead != ChannelManager.HistoryRead.RETRYING && + historyRead != ChannelManager.HistoryRead.READING Column( // The list below is declared after this and would otherwise // sit on top, swallowing taps meant for the button here. @@ -992,10 +1007,10 @@ fun ChatScreen(vm: AppViewModel) { fontSize = 14.sp ) } - // A refused or failed read also clears hasMoreHistory, - // and there the start is unknown, not reached. + // A refused, failed or unfinished read also leaves + // hasMoreHistory false, and there the start is unknown. !hasMoreHistory && historyError == null && visible.isNotEmpty() && - historyRead != ChannelManager.HistoryRead.FAILED -> Box( + historyRead == ChannelManager.HistoryRead.OK -> Box( Modifier.fillMaxWidth().padding(vertical = 16.dp), contentAlignment = Alignment.Center ) {