Concurrent feed fetching with durable fetch-result tracking

Feeds are fetched with bounded concurrency (semaphore of 8) instead of
strictly sequentially -- with 100+ subscriptions, one slow/stalling
server no longer delays every feed queued behind it. A process-local
mutex prevents the periodic and manual/alarm-triggered one-time
fetches (two different WorkManager unique-work names, so its own dedup
doesn't cover this) from ever running concurrently and racing writes
to the same feed rows.

Settings.recordFetchResult persists lastFetchAt/lastFetchNewArticles
on every run, not derived from WorkInfo.State -- a PeriodicWorkRequest
cycles back to ENQUEUED after every successful run and never reaches
SUCCEEDED, so anything watching for that only ever saw manual fetches.
ArticleListScreen surfaces this as a "N new articles" snackbar
(acknowledged durably, so returning to the screen doesn't re-show it)
and computes live "Checking feed X of Y" progress from the worker's
tagged WorkInfo, scoped to the Unread/Older views that actually
aggregate every subscription.

1 files changed, 110 insertions, 26 deletions

diff --git a/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt b/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt
index 5ec6bab..9c2962f 100644
--- a/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt
+++ b/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt
@@ -7,13 +7,24 @@ import androidx.work.NetworkType
 import androidx.work.PeriodicWorkRequestBuilder
 import androidx.work.WorkManager
 import androidx.work.WorkerParameters
+import androidx.work.workDataOf
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withPermit
 import org.terst.feedreader.data.AppDatabase
 import org.terst.feedreader.data.ArticleEntity
 import org.terst.feedreader.data.FeedEntity
 import org.terst.feedreader.data.FeedParser
 import org.terst.feedreader.data.Net
+import org.terst.feedreader.data.Settings
+import org.terst.feedreader.data.await
 import okhttp3.Request
 import java.security.MessageDigest
+import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.TimeUnit
 
 /**
@@ -27,33 +38,80 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor
     private val client = Net.client
 
     override suspend fun doWork(): Result {
-        val db = AppDatabase.get(applicationContext)
-        val feeds = db.feedDao().getAll()
+        // WorkManager's "unique work name" dedup does NOT prevent this: schedulePeriodic
+        // and fetchNow use two different unique-work names (periodic and one-time requests
+        // can't share one), so an hourly periodic run and a pull-to-refresh/morning-alarm
+        // fetch firing at the same moment both start doWork() concurrently -- every feed
+        // gets fetched twice, and both instances race writes to the same feed rows'
+        // etag/lastModified/lastFetchedAt. This mutex is process-local, which is exactly
+        // right here: WorkManager runs this worker in the app's own process, never a
+        // separate one, so a single in-memory lock covers every real overlap case.
+        if (!fetchMutex.tryLock()) {
+            return Result.success()
+        }
+        try {
+            val db = AppDatabase.get(applicationContext)
+            val feeds = db.feedDao().getAll()
+            val completedCount = AtomicInteger(0)
 
-        for (feed in feeds) {
-            fetchOne(feed, db)
+            // Fetched with bounded concurrency, not one feed at a time: with a large
+            // subscription list (this app has seen 100+ feeds in the wild -- see
+            // AppDatabase's v9 migration comments), a strictly sequential loop means one
+            // slow/stalling server delays every feed queued behind it, and the whole run
+            // can take tens of seconds to minutes. Dispatchers.IO (not the CoroutineWorker
+            // default of Dispatchers.Default) because this is blocking network I/O, not
+            // CPU-bound work -- running it on Default would compete with the app's other
+            // CPU-bound coroutine work for that dispatcher's small, core-sized thread pool.
+            val newArticleCount = coroutineScope {
+                val semaphore = Semaphore(FETCH_CONCURRENCY)
+                feeds.map { feed ->
+                    async(Dispatchers.IO) {
+                        semaphore.withPermit {
+                            val count = fetchOne(feed, db)
+                            // KEY_CURRENT now means "how many feeds have finished," not
+                            // "which one's currently being checked" -- there's no single
+                            // "current" feed anymore under concurrent fetching, but a
+                            // monotonically-advancing completed-count is just as good a
+                            // progress signal for the UI.
+                            setProgress(workDataOf(KEY_CURRENT to completedCount.incrementAndGet(), KEY_TOTAL to feeds.size))
+                            count
+                        }
+                    }
+                }.awaitAll().sum()
+            }
+            // Durable, not WorkInfo-state-based -- see Settings.recordFetchResult's doc
+            // comment for why periodic work's own state can't be used for this.
+            Settings(applicationContext).recordFetchResult(newArticleCount)
+            return Result.success(workDataOf(KEY_TOTAL to feeds.size, KEY_NEW_ARTICLES to newArticleCount))
+        } finally {
+            fetchMutex.unlock()
         }
-        return Result.success()
     }
 
-    private suspend fun fetchOne(feed: FeedEntity, db: AppDatabase) {
+    /** Returns how many articles from this feed were genuinely new (not already stored). */
+    private suspend fun fetchOne(feed: FeedEntity, db: AppDatabase): Int {
         val requestBuilder = Request.Builder().url(feed.id)
         feed.etag?.let { requestBuilder.header("If-None-Match", it) }
         feed.lastModified?.let { requestBuilder.header("If-Modified-Since", it) }
 
-        try {
-            client.newCall(requestBuilder.build()).execute().use { response ->
+        return try {
+            client.newCall(requestBuilder.build()).await().use { response ->
                 val now = System.currentTimeMillis()
                 if (response.code == 304) {
-                    db.feedDao().update(feed.copy(lastFetchedAt = now, lastError = null))
-                    return
+                    db.feedDao().updateFetchResult(
+                        feed.id, feed.title, feed.siteUrl, now, feed.etag, feed.lastModified, lastError = null
+                    )
+                    return@use 0
                 }
                 if (!response.isSuccessful) {
-                    db.feedDao().update(feed.copy(lastFetchedAt = now, lastError = "HTTP ${response.code}"))
-                    return
+                    db.feedDao().updateFetchResult(
+                        feed.id, feed.title, feed.siteUrl, now, feed.etag, feed.lastModified,
+                        lastError = "HTTP ${response.code}"
+                    )
+                    return@use 0
                 }
 
-                val body = response.body ?: return
+                val body = response.body ?: return@use 0
                 val parsed = FeedParser.parse(body.byteStream())
 
                 val articles = parsed.items.map { item ->
@@ -70,21 +128,28 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor
                         createdAt = now
                     )
                 }
-                db.articleDao().insertNew(articles)
-
-                db.feedDao().update(
-                    feed.copy(
-                        title = parsed.title.ifBlank { feed.title },
-                        siteUrl = parsed.siteUrl ?: feed.siteUrl,
-                        lastFetchedAt = now,
-                        etag = response.header("ETag") ?: feed.etag,
-                        lastModified = response.header("Last-Modified") ?: feed.lastModified,
-                        lastError = null
-                    )
+                // Room returns -1 for a row OnConflictStrategy.IGNORE actually skipped
+                // (already stored), and the real inserted rowid (>= 0) for a genuinely new
+                // one -- count only the latter so "N new articles" reflects reality.
+                val newCount = db.articleDao().insertNew(articles).count { it != -1L }
+
+                db.feedDao().updateFetchResult(
+                    id = feed.id,
+                    title = parsed.title.ifBlank { feed.title },
+                    siteUrl = parsed.siteUrl ?: feed.siteUrl,
+                    lastFetchedAt = now,
+                    etag = response.header("ETag") ?: feed.etag,
+                    lastModified = response.header("Last-Modified") ?: feed.lastModified,
+                    lastError = null
                 )
+                newCount
             }
         } catch (e: Exception) {
-            db.feedDao().update(feed.copy(lastFetchedAt = System.currentTimeMillis(), lastError = e.message ?: "Fetch failed"))
+            db.feedDao().updateFetchResult(
+                feed.id, feed.title, feed.siteUrl, System.currentTimeMillis(), feed.etag, feed.lastModified,
+                lastError = e.message ?: "Fetch failed"
+            )
+            0
         }
     }
 
@@ -96,6 +161,22 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor
     companion object {
         private const val UNIQUE_WORK_NAME = "feed_fetch_periodic"
 
+        // Common tag on every enqueue (periodic AND one-off) so ArticleListScreen can
+        // observe "is a fetch happening right now" via a single getWorkInfosByTagFlow
+        // query, regardless of which of the two triggered it.
+        const val FETCH_WORK_TAG = "feed_fetch"
+        const val KEY_CURRENT = "current"
+        const val KEY_TOTAL = "total"
+        const val KEY_NEW_ARTICLES = "newArticles"
+
+        // Bounded, not unbounded (Dispatchers.IO can grow to 64+ threads on demand) --
+        // enough to meaningfully overlap slow feeds without opening 100+ simultaneous
+        // connections to 100+ different hosts at once on a fetch run with a large
+        // subscription list.
+        private const val FETCH_CONCURRENCY = 8
+
+        private val fetchMutex = Mutex()
+
         fun schedulePeriodic(context: Context) {
             val request = PeriodicWorkRequestBuilder<FeedFetchWorker>(60, TimeUnit.MINUTES)
                 .setConstraints(
@@ -103,6 +184,7 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor
                         .setRequiredNetworkType(NetworkType.CONNECTED)
                         .build()
                 )
+                .addTag(FETCH_WORK_TAG)
                 .build()
             WorkManager.getInstance(context).enqueueUniquePeriodicWork(
                 UNIQUE_WORK_NAME,
@@ -113,7 +195,9 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor
 
         /** One-off immediate fetch, e.g. a manual pull-to-refresh. */
         fun fetchNow(context: Context) {
-            val request = androidx.work.OneTimeWorkRequestBuilder<FeedFetchWorker>().build()
+            val request = androidx.work.OneTimeWorkRequestBuilder<FeedFetchWorker>()
+                .addTag(FETCH_WORK_TAG)
+                .build()
             WorkManager.getInstance(context).enqueueUniqueWork(
                 "feed_fetch_now",
                 androidx.work.ExistingWorkPolicy.KEEP,