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.

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<NetworkMode> {
 
 fun networkAvailable(context: Context): Flow<Boolean> {
     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<Boolean> {
         val registered = runCatching { connectivity?.registerDefaultNetworkCallback(callback) }.isSuccess
         push()
         awaitClose { if (registered) runCatching { connectivity?.unregisterNetworkCallback(callback) } }
-    }.distinctUntilChanged()
+    }.conflate()
 }

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

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<Boolean>(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 {