From 5614e1b622e45cc23044e1d25ff734ab52fbbe64 Mon Sep 17 00:00:00 2001 From: Pierroons <97373452+Pierroons@users.noreply.github.com> Date: Fri, 7 Aug 2026 22:29:24 +0200 Subject: [PATCH] fix(search): cache paging data before combining, not after MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Typing in the search bar killed the process on the first keystroke: java.lang.IllegalStateException: Attempt to collect twice from pageEventFlow, which is an illegal operation. A PagingData taken straight from a Pager may only be collected once. combine re-emits the latest value of every source whenever any single one of them changes, and extensionSearchResults always emits twice per query — an empty placeholder, then the results — so the same PagingData reached the collector a second time. cachedIn was already there but sat after the combine, where it received the duplicate rather than preventing it. Moving it onto localSearchResults makes the stream multicast; filter and insertHeaderItem still run on every collection, which is what they are meant to do. Searching looked like it simply returned nothing, which is why the SQL was suspected first — it was never at fault. Two tests come with it, including one that reproduces the exception when the cache is absent, so a green suite cannot go quiet on this again. Co-Authored-By: Claude Opus 5 (1M context) --- .../com/m3u/smartphone/ui/AppViewModel.kt | 19 ++- .../ui/AppViewModelSearchPagingTest.kt | 138 ++++++++++++++++++ 2 files changed, 156 insertions(+), 1 deletion(-) create mode 100644 app/smartphone/src/test/java/com/m3u/smartphone/ui/AppViewModelSearchPagingTest.kt diff --git a/app/smartphone/src/main/java/com/m3u/smartphone/ui/AppViewModel.kt b/app/smartphone/src/main/java/com/m3u/smartphone/ui/AppViewModel.kt index baddda108..0f4865bc1 100644 --- a/app/smartphone/src/main/java/com/m3u/smartphone/ui/AppViewModel.kt +++ b/app/smartphone/src/main/java/com/m3u/smartphone/ui/AppViewModel.kt @@ -88,6 +88,23 @@ class AppViewModel @Inject constructor( } } } + // Cache before the combine below, not after. + // + // A PagingData straight out of a Pager can only ever be collected once. + // combine re-emits the latest value of every source whenever any single + // one of them changes, and extensionSearchResults always emits twice per + // query — an empty placeholder, then the actual results. That hands the + // same PagingData downstream a second time, and Paging throws: + // + // IllegalStateException: Attempt to collect twice from pageEventFlow + // + // The crash killed the process on the first keystroke, which is why + // searching looked like it simply returned nothing. + // + // cachedIn turns the stream into a multicast one that tolerates being + // collected again; the filter and insertHeaderItem below are re-applied + // on each collection, which is exactly what we want. + .cachedIn(viewModelScope) private val extensionSearchResults = searchQueries .mapLatestSearchResults { query -> @@ -112,7 +129,7 @@ class AppViewModel @Inject constructor( item = ChannelWithProgramme(channel = channel, programme = null) ) } - }.cachedIn(viewModelScope) + } private fun refreshProgrammes() { viewModelScope.launch { diff --git a/app/smartphone/src/test/java/com/m3u/smartphone/ui/AppViewModelSearchPagingTest.kt b/app/smartphone/src/test/java/com/m3u/smartphone/ui/AppViewModelSearchPagingTest.kt new file mode 100644 index 000000000..aebe61722 --- /dev/null +++ b/app/smartphone/src/test/java/com/m3u/smartphone/ui/AppViewModelSearchPagingTest.kt @@ -0,0 +1,138 @@ +package com.m3u.smartphone.ui + +import androidx.paging.Pager +import androidx.paging.PagingConfig +import androidx.paging.PagingData +import androidx.paging.PagingDataEvent +import androidx.paging.PagingDataPresenter +import androidx.paging.PagingSource +import androidx.paging.PagingState +import androidx.paging.cachedIn +import androidx.paging.insertHeaderItem +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.collectLatest +import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.yield +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test +import kotlin.coroutines.EmptyCoroutineContext + +/** + * The search screen merges a Pager with extension results through [combine]. + * + * combine re-emits the latest value of every source whenever any single one of + * them changes, so the very same PagingData reaches the collector more than + * once — and a PagingData taken straight from a Pager may only be collected + * once. Without a cachedIn in between, Paging throws and the crash takes the + * whole process down on the first keystroke. + */ +class AppViewModelSearchPagingTest { + @Test + fun `a cached stream tolerates the re-emission combine causes`() = runBlocking { + withTimeout(TEST_TIMEOUT_MILLIS) { + val scope = CoroutineScope(Dispatchers.Unconfined + SupervisorJob()) + val promoted = MutableSharedFlow>(replay = 1) + val presenter = SilentPagingDataPresenter() + + val collection = launch { + combinedSearchResults( + local = pagerFlow().cachedIn(scope), + promoted = promoted, + ).collectLatest(presenter::collectFrom) + } + + // The empty placeholder every search emits before its results land. + promoted.emit(emptyList()) + settle() + assertEquals(listOf("local"), presenter.snapshot().items) + + // The results themselves: same PagingData, second collection. + promoted.emit(listOf("promoted")) + settle() + assertEquals(listOf("promoted", "local"), presenter.snapshot().items) + + collection.cancel() + scope.cancel() + } + } + + @Test + fun `an uncached stream is rejected on that second collection`() = runBlocking { + withTimeout(TEST_TIMEOUT_MILLIS) { + val promoted = MutableSharedFlow>(replay = 1) + val presenter = SilentPagingDataPresenter() + var failure: Throwable? = null + + val collection = launch { + runCatching { + combinedSearchResults( + local = pagerFlow(), + promoted = promoted, + ).collectLatest(presenter::collectFrom) + }.onFailure { failure = it } + } + + promoted.emit(emptyList()) + settle() + promoted.emit(listOf("promoted")) + settle() + + val message = failure?.message.orEmpty() + assertTrue( + "expected Paging to reject the second collection, got: $failure", + message.contains("collect twice", ignoreCase = true), + ) + collection.cancel() + } + } + + /** Mirrors how AppViewModel folds promoted results onto the paged ones. */ + private fun combinedSearchResults( + local: Flow>, + promoted: Flow>, + ): Flow> = combine(local, promoted) { paging, items -> + items.asReversed().fold(paging) { acc, item -> acc.insertHeaderItem(item = item) } + } + + private fun pagerFlow(): Flow> = Pager( + config = PagingConfig(pageSize = 20, enablePlaceholders = false), + ) { + SinglePagePagingSource(listOf("local")) + }.flow + + /** Paging hops between dispatchers; give those hops room to complete. */ + private suspend fun settle() = repeat(SETTLE_PASSES) { yield() } + + private class SinglePagePagingSource( + private val items: List, + ) : PagingSource() { + override fun getRefreshKey(state: PagingState): Int? = null + + override suspend fun load(params: LoadParams): LoadResult = + LoadResult.Page(data = items, prevKey = null, nextKey = null) + } + + /** + * Presents nothing — the point is that collecting happens at all. The base + * class keeps the snapshot up to date on its own. + */ + private class SilentPagingDataPresenter : PagingDataPresenter( + mainContext = EmptyCoroutineContext, + ) { + override suspend fun presentPagingDataEvent(event: PagingDataEvent) = Unit + } + + private companion object { + const val TEST_TIMEOUT_MILLIS = 10_000L + const val SETTLE_PASSES = 50 + } +}