Harden the pending-action queue after review

- Order direct stars against the queue so a successful online unstar is no
  longer overwritten by an older queued star, and serialize all queue
  mutations under one lock.
- Treat a SubsonicException with a negative code as a transport failure:
  queue it in setStar and keep it on flush instead of dropping it.
- Make coalescing atomic, keep the offline collector alive across errors,
  and drain from the persisted offline flag so a queued action flushes on
  a cold start that is already online.
- Give ServerActionSink.flush a default body and drop the no-op overrides.

app/src/main/kotlin/com/subtracks/data/db/PendingActionDao.kt

@@ -3,6 +3,7 @@ package com.subtracks.data.db
 import androidx.room3.Dao
 import androidx.room3.Insert
 import androidx.room3.Query
+import androidx.room3.Transaction
 import com.subtracks.data.model.PendingAction
 
 @Dao
@@ -25,4 +26,10 @@ interface PendingActionDao {
         starType: String,
         targetId: String,
     )
+
+    @Transaction
+    suspend fun replace(action: PendingAction) {
+        if (action.starType != null) clearStar(action.sourceId, action.starType, action.targetId)
+        insert(action)
+    }
 }

app/src/main/kotlin/com/subtracks/data/repo/NetworkServerActionSink.kt

@@ -25,6 +25,4 @@ class NetworkServerActionSink(
         val source = sources.activeMusicSource() ?: throw IllegalStateException("No active server")
         source.setStar(type, id, starred)
     }
-
-    override suspend fun flush() = Unit
 }

app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt

@@ -8,58 +8,54 @@ import com.subtracks.data.source.StarType
 import com.subtracks.data.source.subsonic.SubsonicException
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.drop
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
-import java.io.IOException
 
 class PendingServerActionSink(
     private val delegate: ServerActionSink,
     private val db: SubtracksDatabase,
     private val activeSourceId: suspend () -> Long?,
     private val offline: StateFlow<Boolean>,
+    offlinePreferences: Flow<Boolean>,
     scope: CoroutineScope,
 ) : ServerActionSink {
-    private val flushLock = Mutex()
+    private val lock = Mutex()
 
     init {
         scope.launch {
-            // The first emission is the StateFlow's default, not necessarily the persisted value, and
-            // replaying onto the network must not race a cold start with offline mode on.
-            offline
-                .drop(1)
-                .collect { isOffline -> if (!isOffline) flush() }
+            offlinePreferences.collect { isOffline ->
+                // The persisted value is authoritative, unlike the process flag's default at cold
+                // start, so wait for that flag to agree before replaying.
+                if (!isOffline) {
+                    offline.first { !it }
+                    flush()
+                }
+            }
         }
     }
 
     override suspend fun nowPlaying(songId: String) {
         if (offline.value) return
-        try {
-            delegate.nowPlaying(songId)
-            flush()
-        } catch (cancellation: CancellationException) {
-            throw cancellation
-        } catch (_: Exception) {
-        }
+        if (attempt { delegate.nowPlaying(songId) } == null) flush()
     }
 
     override suspend fun scrobble(
         songId: String,
         time: Long,
     ) {
-        if (offline.value) {
-            queue(PendingActionKind.Scrobble, songId, null, time)
-            return
-        }
-        try {
-            delegate.scrobble(songId, time)
-            flush()
-        } catch (cancellation: CancellationException) {
-            throw cancellation
-        } catch (_: Exception) {
-            queue(PendingActionKind.Scrobble, songId, null, time)
+        lock.withLock {
+            if (offline.value) {
+                enqueueLocked(PendingActionKind.Scrobble, songId, null, time)
+                return
+            }
+            when (val failure = attempt { delegate.scrobble(songId, time) }) {
+                null -> drainSafelyLocked()
+                else -> if (isRetryable(failure)) enqueueLocked(PendingActionKind.Scrobble, songId, null, time)
+            }
         }
     }
 
@@ -69,41 +65,58 @@ class PendingServerActionSink(
         starred: Boolean,
     ) {
         val kind = if (starred) PendingActionKind.Star else PendingActionKind.Unstar
-        if (offline.value) {
-            if (!queue(kind, id, type.name, 0)) throw IllegalStateException("No active server")
-            return
-        }
-        try {
-            delegate.setStar(type, id, starred)
-            flush()
-        } catch (cancellation: CancellationException) {
-            throw cancellation
-        } catch (rejected: SubsonicException) {
-            // The server answered and refused; retrying the same request will not help.
-            throw rejected
-        } catch (failure: Exception) {
-            if (!queue(kind, id, type.name, 0)) throw failure
+        lock.withLock {
+            if (offline.value) {
+                if (!enqueueLocked(kind, id, type.name, 0)) throw IllegalStateException("No active server")
+                return
+            }
+            when (val failure = attempt { delegate.setStar(type, id, starred) }) {
+                null -> {
+                    // The absolute state just sent wins; a queued star for the same item is older.
+                    clearStarLocked(type.name, id)
+                    drainSafelyLocked()
+                }
+
+                else -> {
+                    if (!isRetryable(failure) || !enqueueLocked(kind, id, type.name, 0)) throw failure
+                }
+            }
         }
     }
 
     override suspend fun flush() {
-        flushLock.withLock {
-            if (offline.value) return
-            val sourceId = activeSourceId() ?: return
-            val dao = db.pendingActionDao()
-            for (action in dao.pending(sourceId)) {
-                try {
-                    deliver(action)
-                } catch (cancellation: CancellationException) {
-                    throw cancellation
-                } catch (_: IOException) {
-                    // Still unreachable: keep this and the remaining actions for the next attempt.
-                    return
-                } catch (_: Exception) {
-                    // The server answered and refused; drop the action rather than block the queue.
-                }
-                dao.delete(action.id)
+        try {
+            lock.withLock { drainLocked() }
+        } catch (cancellation: CancellationException) {
+            throw cancellation
+        } catch (_: Exception) {
+        }
+    }
+
+    private suspend fun drainLocked() {
+        if (offline.value) return
+        val sourceId = activeSourceId() ?: return
+        val dao = db.pendingActionDao()
+        for (action in dao.pending(sourceId)) {
+            try {
+                deliver(action)
+            } catch (cancellation: CancellationException) {
+                throw cancellation
+            } catch (failure: Exception) {
+                // Still unreachable: keep this and the remaining actions for the next attempt.
+                if (isRetryable(failure)) return
+                // The server answered and refused; drop the action rather than block the queue.
             }
+            dao.delete(action.id)
+        }
+    }
+
+    private suspend fun drainSafelyLocked() {
+        try {
+            drainLocked()
+        } catch (cancellation: CancellationException) {
+            throw cancellation
+        } catch (_: Exception) {
         }
     }
 
@@ -120,16 +133,38 @@ class PendingServerActionSink(
         }
     }
 
-    private suspend fun queue(
+    private suspend fun enqueueLocked(
         kind: PendingActionKind,
         targetId: String,
         starType: String?,
         time: Long,
     ): Boolean {
         val sourceId = activeSourceId() ?: return false
-        val dao = db.pendingActionDao()
-        if (starType != null) dao.clearStar(sourceId, starType, targetId)
-        dao.insert(PendingAction(sourceId = sourceId, kind = kind, targetId = targetId, starType = starType, time = time))
+        db.pendingActionDao().replace(
+            PendingAction(sourceId = sourceId, kind = kind, targetId = targetId, starType = starType, time = time),
+        )
         return true
     }
+
+    private suspend fun clearStarLocked(
+        starType: String,
+        targetId: String,
+    ) {
+        val sourceId = activeSourceId() ?: return
+        db.pendingActionDao().clearStar(sourceId, starType, targetId)
+    }
+
+    private suspend fun attempt(block: suspend () -> Unit): Exception? =
+        try {
+            block()
+            null
+        } catch (cancellation: CancellationException) {
+            throw cancellation
+        } catch (failure: Exception) {
+            failure
+        }
+
+    // A SubsonicException with a negative code is a transport or protocol failure (HTTP non-2xx,
+    // malformed body), not a refusal, so it is still worth retrying.
+    private fun isRetryable(failure: Exception): Boolean = failure !is SubsonicException || failure.code < 0
 }

app/src/main/kotlin/com/subtracks/data/source/ServerActionSink.kt

@@ -18,5 +18,5 @@ interface ServerActionSink {
         starred: Boolean,
     )
 
-    suspend fun flush()
+    suspend fun flush() = Unit
 }

app/src/main/kotlin/com/subtracks/di/AppModule.kt

@@ -84,6 +84,7 @@ fun appModule(
             db = get(),
             activeSourceId = get<SourceRepository>()::activeSourceIdOnce,
             offline = get<SourceRepository>().offline,
+            offlinePreferences = get<UserPreferences>().offlineMode(),
             scope = get(),
         )
     }

app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt

@@ -5,6 +5,7 @@ import androidx.room3.Room
 import androidx.sqlite.driver.bundled.BundledSQLiteDriver
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.subtracks.awaitUntil
 import com.subtracks.cancelAndJoinBlocking
 import com.subtracks.data.db.SubtracksDatabase
 import com.subtracks.data.model.PendingActionKind
@@ -64,8 +65,6 @@ class PendingServerActionSinkTest {
             starFailure?.let { throw it }
             calls += Call.Star(type, id, starred)
         }
-
-        override suspend fun flush() = Unit
     }
 
     private lateinit var db: SubtracksDatabase
@@ -73,6 +72,7 @@ class PendingServerActionSinkTest {
     private lateinit var delegate: RecordingSink
     private lateinit var sink: PendingServerActionSink
     private val offline = MutableStateFlow(true)
+    private val offlinePreferences = MutableStateFlow(true)
     private var sourceId = 0L
 
     @Before
@@ -86,7 +86,15 @@ class PendingServerActionSinkTest {
         scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
         sourceId = runBlocking { db.sourcesDao().insertSource(Source(name = "server", address = "http://x/", createdAt = 0)) }
         delegate = RecordingSink()
-        sink = PendingServerActionSink(delegate, db, activeSourceId = { sourceId }, offline = offline, scope = scope)
+        sink =
+            PendingServerActionSink(
+                delegate = delegate,
+                db = db,
+                activeSourceId = { sourceId },
+                offline = offline,
+                offlinePreferences = offlinePreferences,
+                scope = scope,
+            )
     }
 
     @After
@@ -110,6 +118,19 @@ class PendingServerActionSinkTest {
             assertEquals(0, db.pendingActionDao().pending(sourceId).size)
         }
 
+    @Test
+    fun turningOfflineOffReplaysThroughTheCollector() =
+        runBlocking {
+            sink.setStar(StarType.Song, "s1", true)
+
+            offline.value = false
+            offlinePreferences.value = false
+
+            awaitUntil("the queued star to replay") { delegate.calls.isNotEmpty() }
+            assertEquals(listOf(Call.Star(StarType.Song, "s1", true)), delegate.calls)
+            assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+        }
+
     @Test
     fun anOfflineUnstarCoalescesAwayAnEarlierStar() =
         runBlocking {
@@ -126,6 +147,21 @@ class PendingServerActionSinkTest {
             assertEquals(listOf(Call.Star(StarType.Song, "s1", false)), delegate.calls)
         }
 
+    @Test
+    fun aQueuedStarCannotOverwriteANewerOnlineUnstar() =
+        runBlocking {
+            offline.value = false
+            delegate.starFailure = IOException("down")
+            sink.setStar(StarType.Song, "s1", true)
+            assertEquals(1, db.pendingActionDao().pending(sourceId).size)
+
+            delegate.starFailure = null
+            sink.setStar(StarType.Song, "s1", false)
+
+            assertEquals(listOf(Call.Star(StarType.Song, "s1", false)), delegate.calls)
+            assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+        }
+
     @Test
     fun offlineScrobblesReplayInOrderWithTheirTimes() =
         runBlocking {
@@ -169,6 +205,17 @@ class PendingServerActionSinkTest {
             assertEquals(0, db.pendingActionDao().pending(sourceId).size)
         }
 
+    @Test
+    fun aTransportFailureIsQueuedRatherThanRejected() =
+        runBlocking {
+            offline.value = false
+            delegate.starFailure = SubsonicException(-1, "HTTP 503")
+
+            sink.setStar(StarType.Song, "s1", true)
+
+            assertEquals(1, db.pendingActionDao().pending(sourceId).size)
+        }
+
     @Test
     fun flushKeepsQueuedActionsWhileTheServerIsStillUnreachable() =
         runBlocking {
@@ -186,4 +233,22 @@ class PendingServerActionSinkTest {
             assertEquals(listOf(Call.Star(StarType.Song, "s1", true)), delegate.calls)
             assertEquals(0, db.pendingActionDao().pending(sourceId).size)
         }
+
+    @Test
+    fun flushKeepsQueuedActionsOnATransportFailure() =
+        runBlocking {
+            sink.scrobble("s1", 1_000)
+
+            offline.value = false
+            delegate.scrobbleFailure = SubsonicException(-1, "HTTP 503")
+            sink.flush()
+
+            assertEquals(1, db.pendingActionDao().pending(sourceId).size)
+
+            delegate.scrobbleFailure = null
+            sink.flush()
+
+            assertEquals(listOf(Call.Scrobble("s1", 1_000)), delegate.calls)
+            assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+        }
 }

app/src/test/kotlin/com/subtracks/playback/ScrobblerTest.kt

@@ -48,8 +48,6 @@ class ScrobblerTest {
             id: String,
             starred: Boolean,
         ) = Unit
-
-        override suspend fun flush() = Unit
     }
 
     private fun TestScope.testScope() = CoroutineScope(UnconfinedTestDispatcher(testScheduler) + SupervisorJob())