Refresh skips with bounded concurrency, not one at a time

The sweep awaited each episode's server query sequentially, and it runs on
every refresh including the library button. Several hundred downloaded
episodes meant a tap that used to take seconds took minutes, and under
WorkManager it risked being killed partway -- the same shape as the
ten-minute ceiling that stopped FingerprintWorker ever confirming.

Six at a time, through a semaphore inside the call's own coroutineScope so
nothing escapes to a background dispatcher. Per-episode failure isolation
is unchanged; CancellationException is rethrown rather than caught, the
bug already fixed once in skipFurniture().

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018AwnALZL6bD4VBtATDVv3B

2 files changed, 150 insertions, 6 deletions

diff --git a/app/src/main/java/org/terst/listen/server/SkipRefresher.kt b/app/src/main/java/org/terst/listen/server/SkipRefresher.kt
index d2cf4fe..4d490f6 100644
--- a/app/src/main/java/org/terst/listen/server/SkipRefresher.kt
+++ b/app/src/main/java/org/terst/listen/server/SkipRefresher.kt
@@ -1,9 +1,25 @@
 package org.terst.listen.server
 
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withPermit
 import org.terst.listen.data.EpisodeDao
 import org.terst.listen.download.ServerAudio
 
 /**
+ * How many episodes' server queries run at once.
+ *
+ * Peter's own small Go server, not a CDN: this is sized to get the benefit of overlapping
+ * round trips without hammering a single-process backend with hundreds of simultaneous
+ * requests. Kept at the low end of "a small fixed limit" on purpose -- there is no reason to
+ * push it higher just because the library is bigger.
+ */
+private const val CONCURRENT_QUERIES = 6
+
+/**
  * Picks up improved skip detection on a feed refresh, for the episodes it is safe to re-query.
  *
  * `EpisodeDownloader` only asks the server once, at download time, so an episode's offsets are
@@ -25,15 +41,40 @@ class SkipRefresher(
 ) {
 
     /**
+     * Runs every eligible episode's query with bounded concurrency rather than one at a time --
+     * a real library runs to several hundred downloaded episodes, and sequentially awaiting
+     * each one made a single refresh-button tap take minutes instead of seconds (the same class
+     * of problem `FingerprintWorker` hit against WorkManager's ten-minute ceiling).
+     *
      * One episode's failure -- the server unreachable, a malformed response -- must not stop
      * the rest of the library from being refreshed, the same reasoning as every other per-item
-     * loop in this app.
+     * loop in this app. [CancellationException] is the one throwable that must *not* be caught
+     * here: swallowing it would let a cancelled refresh silently keep running, the exact bug
+     * already fixed once in `skipFurniture()`.
+     *
+     * `coroutineScope` keeps every launched query a structured child of this call -- nothing
+     * escapes to an outer scope or a background dispatcher, and cancelling the caller cancels
+     * every query still in flight.
      */
-    suspend fun refreshAll() {
-        episodes.downloadedFromServer().forEach { episode ->
-            runCatching {
-                serverAudio.resolve(episode)?.let { info -> serverAudio.record(episode, info) }
+    suspend fun refreshAll() = coroutineScope {
+        val semaphore = Semaphore(CONCURRENT_QUERIES)
+        episodes.downloadedFromServer().map { episode ->
+            async {
+                semaphore.withPermit {
+                    // Deliberately broad. One episode's failure -- the server unreachable, a
+                    // malformed response -- must not stop the rest of the sweep, so it is
+                    // isolated here rather than left to propagate. Logged rather than silently
+                    // discarded, same reasoning as skipFurniture()'s catch just above.
+                    @Suppress("TooGenericExceptionCaught")
+                    try {
+                        serverAudio.resolve(episode)?.let { info -> serverAudio.record(episode, info) }
+                    } catch (e: CancellationException) {
+                        throw e
+                    } catch (e: Exception) {
+                        android.util.Log.w("SkipRefresher", "skip refresh failed for ${episode.id}", e)
+                    }
+                }
             }
-        }
+        }.awaitAll()
     }
 }
diff --git a/app/src/test/java/org/terst/listen/server/SkipRefresherTest.kt b/app/src/test/java/org/terst/listen/server/SkipRefresherTest.kt
index 9dffdb8..1858120 100644
--- a/app/src/test/java/org/terst/listen/server/SkipRefresherTest.kt
+++ b/app/src/test/java/org/terst/listen/server/SkipRefresherTest.kt
@@ -1,5 +1,10 @@
 package org.terst.listen.server
 
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
@@ -119,6 +124,104 @@ class SkipRefresherTest {
     }
 
     @Test
+    fun `the whole library is refreshed, not just the first batch`() = runTest {
+        // More episodes than the concurrency limit, so this only passes if the sweep keeps
+        // going past the first bounded batch rather than silently dropping the rest.
+        val ids = (1..25).map { "e$it" }
+        val dao = FakeEpisodeDao().apply { ids.forEach { rows += episode(it) } }
+        val audio = FakeServerAudio(infoFor = ids.associateWith { info(1_000) })
+
+        SkipRefresher(dao, audio).refreshAll()
+
+        assertEquals(ids.toSet(), audio.resolved.toSet())
+        assertEquals(ids.size, audio.recorded.size)
+    }
+
+    @Test
+    fun `one bad episode among many concurrent ones does not stop the rest`() = runTest {
+        // Enough episodes to span several concurrency batches, with failures interleaved
+        // throughout rather than confined to one batch, so a regression that only isolates
+        // failures within a single batch would still be caught.
+        val ids = (1..25).map { "e$it" }
+        val failing = ids.filterIndexed { index, _ -> index % 3 == 0 }.toSet()
+        val succeeding = ids - failing
+        val dao = FakeEpisodeDao().apply { ids.forEach { rows += episode(it) } }
+        val audio = FakeServerAudio(
+            infoFor = succeeding.associateWith { info(2_000) },
+            failing = failing,
+        )
+
+        SkipRefresher(dao, audio).refreshAll()
+
+        assertEquals(ids.toSet(), audio.resolved.toSet())
+        assertEquals(succeeding.toSet(), audio.recorded.map { it.first }.toSet())
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `cancelling the sweep cancels in-flight queries rather than letting them finish`() = runTest {
+        val ids = (1..10).map { "e$it" }
+        val dao = FakeEpisodeDao().apply { ids.forEach { rows += episode(it) } }
+
+        // Blocks forever until cancelled, so a test that lets this complete normally proves
+        // the sweep waited for the cancelled work instead of abandoning it.
+        val stuck = CompletableDeferred<ServerEpisode?>()
+        val started = mutableListOf<String>()
+        val audio = object : ServerAudio {
+            override suspend fun resolve(episode: EpisodeEntity): ServerEpisode? {
+                started += episode.id
+                return stuck.await()
+            }
+
+            override suspend fun record(episode: EpisodeEntity, info: ServerEpisode) {
+                error("must never be reached: resolve() never returns")
+            }
+        }
+
+        val job = launch { SkipRefresher(dao, audio).refreshAll() }
+        runCurrent()
+        assertTrue(
+            "several queries should run at once rather than strictly one at a time",
+            started.size > 1,
+        )
+        assertTrue(
+            "a bounded sweep must not start every episode's query at once",
+            started.size < ids.size,
+        )
+
+        job.cancel()
+        job.join()
+
+        assertTrue(job.isCancelled)
+    }
+
+    @Test
+    fun `a per-episode failure must not swallow cancellation`() = runTest {
+        // Guards against the runCatching(Throwable)-swallows-CancellationException bug this
+        // codebase already fixed once (skipFurniture()): a per-episode catch that is too broad
+        // would eat the cancellation here and let refreshAll() return normally instead of
+        // propagating it.
+        val dao = FakeEpisodeDao().apply { rows += episode("e1") }
+        val audio = object : ServerAudio {
+            override suspend fun resolve(episode: EpisodeEntity): ServerEpisode? {
+                throw CancellationException("cancelled mid-query")
+            }
+
+            override suspend fun record(episode: EpisodeEntity, info: ServerEpisode) {
+                error("must never be reached")
+            }
+        }
+
+        val job = launch { SkipRefresher(dao, audio).refreshAll() }
+        job.join()
+
+        assertTrue(
+            "CancellationException from a single episode must propagate, not be swallowed",
+            job.isCancelled,
+        )
+    }
+
+    @Test
     fun `an episode the server no longer has anything to say about is left alone`() = runTest {
         val dao = FakeEpisodeDao().apply { rows += episode("e1") }
         val audio = FakeServerAudio(infoFor = mapOf("e1" to null))