Fix connectivity listener

Fixes an issue where another VPN app or user having unfortunate
timing of turning on airplane mode and connecting at almost the same
time would leave a lingering network cached in the scan. This fix will
start with the all networks state and give the networkEvents flow 1
second to start up and emit the actual network state.

2 files changed, 72 insertions, 27 deletions

diff --git a/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt b/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt
index bf94c80778..863fc32251 100644
--- a/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt
+++ b/android/lib/common/src/main/kotlin/net/mullvad/mullvadvpn/lib/common/util/CommonFlowUtils.kt
@@ -1,9 +1,26 @@
 package net.mullvad.mullvadvpn.lib.common.util
 
+import kotlin.time.Duration
+import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.debounce
 import kotlinx.coroutines.flow.firstOrNull
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.withIndex
 import kotlinx.coroutines.withTimeoutOrNull
 
 suspend fun <T> Flow<T>.firstOrNullWithTimeout(timeMillis: Long): T? {
     return withTimeoutOrNull(timeMillis) { firstOrNull() }
 }
+
+@OptIn(FlowPreview::class)
+fun <T> Flow<T>.debounceFirst(timeout: Duration): Flow<T> =
+    withIndex()
+        .debounce {
+            if (it.index == 0) {
+                timeout
+            } else {
+                Duration.ZERO
+            }
+        }
+        .map { it.value }
diff --git a/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt b/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt
index 6e334a9ead..3505e4496b 100644
--- a/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt
+++ b/android/lib/talpid/src/main/kotlin/net/mullvad/talpid/ConnectivityListener.kt
@@ -2,11 +2,13 @@ package net.mullvad.talpid
 
 import android.net.ConnectivityManager
 import android.net.LinkProperties
+import android.net.Network
 import android.net.NetworkCapabilities
 import android.net.NetworkRequest
 import co.touchlab.kermit.Logger
 import java.net.InetAddress
 import kotlin.collections.ArrayList
+import kotlin.time.Duration.Companion.seconds
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
@@ -14,13 +16,16 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.flow.scan
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
+import net.mullvad.mullvadvpn.lib.common.util.debounceFirst
 import net.mullvad.talpid.model.NetworkState
 import net.mullvad.talpid.util.NetworkEvent
 import net.mullvad.talpid.util.RawNetworkState
@@ -59,7 +64,7 @@ class ConnectivityListener(private val connectivityManager: ConnectivityManager)
         }
 
         _isConnected =
-            hasInternetCapability()
+            hasInternetConnectivity()
                 .onEach { notifyConnectivityChange(it) }
                 .stateIn(
                     scope,
@@ -79,40 +84,63 @@ class ConnectivityListener(private val connectivityManager: ConnectivityManager)
     private fun LinkProperties.dnsServersWithoutFallback(): List<InetAddress> =
         dnsServers.filter { it.hostAddress != TalpidVpnService.FALLBACK_DUMMY_DNS_SERVER }
 
-    private val nonVPNNetworksRequest =
-        NetworkRequest.Builder().addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN).build()
+    private val nonVPNInternetNetworksRequest =
+        NetworkRequest.Builder()
+            .addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
+            .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+            .build()
 
-    private fun hasInternetCapability(): Flow<Boolean> {
-        @Suppress("DEPRECATION")
+    /**
+     * Return a flow notifying us if we have internet connectivity. Initial state will be taken from
+     * `allNetworks` and then updated when network events occur. Important to note that
+     * `allNetworks` may return a network that we never get updates from if turned off at the moment
+     * of the initial query.
+     */
+    private fun hasInternetConnectivity(): Flow<Boolean> {
         return connectivityManager
-            .networkEvents(nonVPNNetworksRequest)
-            .scan(
-                connectivityManager.allNetworks.associateWith {
-                    connectivityManager.getNetworkCapabilities(it)
-                }
-            ) { networks, event ->
+            .networkEvents(nonVPNInternetNetworksRequest)
+            .filter { it is NetworkEvent.Lost || it is NetworkEvent.CapabilitiesChanged }
+            .scan(emptySet<Network>()) { networks, event ->
                 when (event) {
-                    is NetworkEvent.Lost -> {
-                        Logger.d("Network lost ${event.network}")
-                        (networks - event.network).also {
-                            Logger.d("Number of networks: ${it.size}")
-                        }
-                    }
-                    is NetworkEvent.CapabilitiesChanged -> {
-                        Logger.d("Network capabilities changed ${event.network}")
-                        (networks + (event.network to event.networkCapabilities)).also {
-                            Logger.d("Number of networks: ${it.size}")
-                        }
+                    is NetworkEvent.Lost -> networks - event.network
+                    is NetworkEvent.Available -> networks + event.network
+                    else -> networks // Should never happen
+                }.also { Logger.d("Networks: $it") }
+            }
+            // NetworkEvents are slow, can several 100 millis to arrive. If we are online, we don't
+            // want to emit a false offline with the initial accumulator, so we wait a bit before
+            // emitting, and rely on `networksWithInternetConnectivity`.
+            //
+            // Also if our initial state was "online", but it just got turned off we might not see
+            // any updates for this network even though we already were registered for updated, and
+            // thus we can't drop initial value accumulator value.
+            .debounceFirst(1.seconds)
+            .onStart {
+                // We should not use this as initial state in scan, because it may contain networks
+                // that won't be included in `networkEvents` updates.
+                emit(
+                    connectivityManager.networksWithInternetConnectivity().also {
+                        Logger.d("Networks (Initial): $it")
                     }
-                    else -> networks
-                }
+                )
             }
-            .map { it.any { it.value.hasInternetCapability() } }
+            .map { it.isNotEmpty() }
             .distinctUntilChanged()
     }
 
-    private fun NetworkCapabilities?.hasInternetCapability(): Boolean =
-        this?.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) == true
+    @Suppress("DEPRECATION")
+    private fun ConnectivityManager.networksWithInternetConnectivity(): Set<Network> =
+        // Currently the use of `allNetworks` (which is deprecated in favor of listening to network
+        // events) is our only option because network events does not give us the initial state fast
+        // enough.
+        allNetworks
+            .filter {
+                val capabilities = getNetworkCapabilities(it) ?: return@filter false
+
+                capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) &&
+                    capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
+            }
+            .toSet()
 
     private fun RawNetworkState.toNetworkState(): NetworkState =
         NetworkState(