From f20937c83e0642ae80c9d6f5461b25559396f942 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Sat, 3 Oct 2026 23:51:03 +0000 Subject: [PATCH] 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. --- .../com/subtracks/data/db/PendingActionDao.kt | 7 + .../data/repo/NetworkServerActionSink.kt | 2 - .../data/repo/PendingServerActionSink.kt | 155 +++++++++++------- .../subtracks/data/source/ServerActionSink.kt | 2 +- .../main/kotlin/com/subtracks/di/AppModule.kt | 1 + .../data/repo/PendingServerActionSinkTest.kt | 71 +++++++- .../com/subtracks/playback/ScrobblerTest.kt | 2 - 7 files changed, 172 insertions(+), 68 deletions(-) diff --git a/app/src/main/kotlin/com/subtracks/data/db/PendingActionDao.kt b/app/src/main/kotlin/com/subtracks/data/db/PendingActionDao.kt index 42e02239..fd72ef43 100644 --- a/app/src/main/kotlin/com/subtracks/data/db/PendingActionDao.kt +++ b/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) + } } diff --git a/app/src/main/kotlin/com/subtracks/data/repo/NetworkServerActionSink.kt b/app/src/main/kotlin/com/subtracks/data/repo/NetworkServerActionSink.kt index 3e134f3e..dc94c4ca 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/NetworkServerActionSink.kt +++ b/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 } diff --git a/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt b/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt index a1e3bbd3..aa06b67f 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt +++ b/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, + offlinePreferences: Flow, 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 } diff --git a/app/src/main/kotlin/com/subtracks/data/source/ServerActionSink.kt b/app/src/main/kotlin/com/subtracks/data/source/ServerActionSink.kt index 26a8fd3e..0af6369c 100644 --- a/app/src/main/kotlin/com/subtracks/data/source/ServerActionSink.kt +++ b/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 } diff --git a/app/src/main/kotlin/com/subtracks/di/AppModule.kt b/app/src/main/kotlin/com/subtracks/di/AppModule.kt index ff191606..4f94f3f2 100644 --- a/app/src/main/kotlin/com/subtracks/di/AppModule.kt +++ b/app/src/main/kotlin/com/subtracks/di/AppModule.kt @@ -84,6 +84,7 @@ fun appModule( db = get(), activeSourceId = get()::activeSourceIdOnce, offline = get().offline, + offlinePreferences = get().offlineMode(), scope = get(), ) } diff --git a/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt b/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt index d8e7829a..1608bd82 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt +++ b/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) + } } diff --git a/app/src/test/kotlin/com/subtracks/playback/ScrobblerTest.kt b/app/src/test/kotlin/com/subtracks/playback/ScrobblerTest.kt index 7aa0af54..4043d31e 100644 --- a/app/src/test/kotlin/com/subtracks/playback/ScrobblerTest.kt +++ b/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())