From 9f2e14a9f982be6d67d32d268a1bed65a9488747 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Sun, 4 Oct 2026 00:24:42 +0000 Subject: [PATCH] Re-notify replay on every connectivity change, not just availability flips conflate rather than distinctUntilChanged: switching between two available networks (WiFi LAN server -> cellular -> WiFi) kept the signal true, so the queue never drained until another trigger. Emit on each callback instead, and drop the sink combine's distinctUntilChanged so a repeated ready edge still flushes. Simplify flush to reuse the safe drain. Add a regression test that replays on two true emissions with no false between. --- .../com/subtracks/data/net/NetworkMode.kt | 5 ++- .../data/repo/PendingServerActionSink.kt | 11 +----- .../data/repo/PendingServerActionSinkTest.kt | 39 +++++++++++++++++++ 3 files changed, 44 insertions(+), 11 deletions(-) diff --git a/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt b/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt index 9f315906..271b2587 100644 --- a/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt +++ b/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt @@ -7,6 +7,7 @@ import android.net.NetworkCapabilities import kotlinx.coroutines.channels.awaitClose import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.callbackFlow +import kotlinx.coroutines.flow.conflate import kotlinx.coroutines.flow.distinctUntilChanged enum class NetworkMode( @@ -55,6 +56,8 @@ fun networkMode(context: Context): Flow { fun networkAvailable(context: Context): Flow { val connectivity = context.applicationContext.getSystemService(ConnectivityManager::class.java) + // Conflated, not distinct: switching between two available networks must still re-notify so the + // replay retries against the (possibly different) newly active route to the server. return callbackFlow { fun push() { val capabilities = @@ -76,5 +79,5 @@ fun networkAvailable(context: Context): Flow { val registered = runCatching { connectivity?.registerDefaultNetworkCallback(callback) }.isSuccess push() awaitClose { if (registered) runCatching { connectivity?.unregisterNetworkCallback(callback) } } - }.distinctUntilChanged() + }.conflate() } 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 a03e8861..49636bf3 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt @@ -11,7 +11,6 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.combine -import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex @@ -31,7 +30,6 @@ class PendingServerActionSink( init { scope.launch { combine(offlinePreferences, networkAvailable) { isOffline, online -> !isOffline && online } - .distinctUntilChanged() .collect { ready -> // The persisted flag and connectivity are authoritative, unlike the process // flag's default at cold start, so wait for that flag to agree before replaying. @@ -89,14 +87,7 @@ class PendingServerActionSink( } } - override suspend fun flush() { - try { - lock.withLock { drainLocked() } - } catch (cancellation: CancellationException) { - throw cancellation - } catch (_: Exception) { - } - } + override suspend fun flush() = lock.withLock { drainSafelyLocked() } private suspend fun drainLocked() { if (offline.value) return 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 8d12b68c..ab2a57fa 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt @@ -17,6 +17,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.runBlocking @@ -170,6 +171,44 @@ class PendingServerActionSinkTest { } } + @OptIn(ExperimentalCoroutinesApi::class) + @Test + fun aNetworkSwitchThatStaysAvailableReplaysAgain() = + runBlocking { + offline.value = false + val network = MutableSharedFlow(replay = 1, extraBufferCapacity = 1) + network.tryEmit(false) + val reconnectScope = CoroutineScope(SupervisorJob() + UnconfinedTestDispatcher()) + try { + val reconnectable = + PendingServerActionSink( + delegate = delegate, + db = db, + activeSourceId = { sourceId }, + offline = offline, + offlinePreferences = flowOf(false), + networkAvailable = network, + scope = reconnectScope, + ) + delegate.scrobbleFailure = IOException("down") + reconnectable.scrobble("s1", 1_000) + delegate.scrobbleFailure = null + network.tryEmit(true) + awaitUntil("the first replay") { delegate.calls.size == 1 } + + delegate.scrobbleFailure = IOException("down") + reconnectable.scrobble("s2", 2_000) + delegate.scrobbleFailure = null + // No false in between: an available-network switch must still re-notify. + network.tryEmit(true) + + awaitUntil("the second replay") { delegate.calls.size == 2 } + assertEquals(listOf(Call.Scrobble("s1", 1_000), Call.Scrobble("s2", 2_000)), delegate.calls) + } finally { + cancelAndJoinBlocking(reconnectScope) + } + } + @Test fun anOfflineUnstarCoalescesAwayAnEarlierStar() = runBlocking {