Fix stale blocking warnings and QUIC failure fan-out

app/src/main/java/net/megaproxy487/vpn/BlockingDetection.kt

@@ -47,3 +47,32 @@ internal fun nativeFailureReason(message: String): String {
         else -> "other"
     }
 }
+
+/** Count independent failures within a bounded window, rather than every failed stream. */
+internal class BlockingEvidence {
+    private var firstFailureAt: Long? = null
+    private var count = 0
+    private val failedSessions = mutableSetOf<String>()
+
+    @Synchronized fun failure(now: Long, detail: String): Int {
+        val first = firstFailureAt
+        if (first == null || now - first > 60_000L) {
+            firstFailureAt = now
+            count = 0
+            failedSessions.clear()
+        }
+        val session = detail.split(' ').firstOrNull { it.startsWith("quic_session=") }
+            ?.removePrefix("quic_session=")?.takeUnless { it == "0" }
+        if (session != null && !failedSessions.add(session)) return 0
+        return ++count
+    }
+
+    @Synchronized fun recovered() {
+        firstFailureAt = null
+        count = 0
+        failedSessions.clear()
+    }
+}
+
+internal fun isProxyTunnelEstablished(message: String): Boolean =
+    "hop=jump" !in message && transportProtocolFromDiagnostic(message)?.let { it != VpnTransportProtocol.SSH_MULTIPLEXED } == true

app/src/main/java/net/megaproxy487/vpn/ProxyCore.kt

@@ -136,7 +136,7 @@ class NativeProxyCore(
             val reporter = reporter { message ->
                 diagnostics(message)
                 VpnRuntimeState.observeDiagnostic(message)
-                if ("SSH_HOST_KEY_" in message || "dpi_hint=possible" in message) status(message)
+                if ("SSH_HOST_KEY_" in message || "dpi_hint=possible" in message || isProxyTunnelEstablished(message)) status(message)
             }
             val json = configJson(config)
             // Java keeps its duplicate alive for the call; Go duplicates it on entry.

app/src/main/java/net/megaproxy487/vpn/ProxyVpnService.kt

@@ -54,8 +54,8 @@ class ProxyVpnService : VpnService() {
     private val tunnelStateLock = Any()
     private var pendingStart: Pair<ParcelFileDescriptor, ProxyCore>? = null
     private val startGeneration = AtomicLong(0)
-    private val probableFailureCounts = ConcurrentHashMap<String, Int>()
-    private val probableFailureTimes = ConcurrentHashMap<String, Long>()
+    private val blockingEvidence = ConcurrentHashMap<String, BlockingEvidence>()
+    @Volatile private var blockingWarning: String? = null
     private val attemptedFailoverProfiles = ConcurrentHashMap.newKeySet<String>()
     private val freshDnsRecoveryProfiles = ConcurrentHashMap.newKeySet<String>()
     private val underlyingCapabilities = ConcurrentHashMap<Network, NetworkCapabilities>()
@@ -114,7 +114,9 @@ class ProxyVpnService : VpnService() {
 
     override fun onCreate() {
         super.onCreate()
-        failoverNotice = ConfigStore(this).failoverNotice()
+        val store = ConfigStore(this)
+        failoverNotice = store.failoverNotice()
+        if (!store.isFailoverActive()) blockingWarning = failoverNotice
         failoverNotice?.let(VpnRuntimeState::updateNetworkWarning)
         createChannel()
         registerUnderlyingNetworkCallback()
@@ -155,7 +157,7 @@ class ProxyVpnService : VpnService() {
             }
         }
         if (intent?.action == ACTION_START_MANUAL) {
-            probableFailureCounts.clear(); probableFailureTimes.clear(); attemptedFailoverProfiles.clear(); failoverNotice = null
+            blockingEvidence.clear(); attemptedFailoverProfiles.clear(); failoverNotice = null; blockingWarning = null
             freshDnsRecoveryProfiles.clear()
             connectionBlockedForAction = false
             resetRetryState()
@@ -179,11 +181,11 @@ class ProxyVpnService : VpnService() {
             return START_NOT_STICKY
         }
         if (intent?.action == ACTION_RECONNECT) {
-            hostKeyPrompt = null; failoverNotice = null
+            hostKeyPrompt = null; failoverNotice = null; blockingWarning = null
             connectionBlockedForAction = false
             resetRetryState()
             store.setFailoverState(false, null)
-            probableFailureCounts.clear(); probableFailureTimes.clear(); attemptedFailoverProfiles.clear()
+            blockingEvidence.clear(); attemptedFailoverProfiles.clear()
             VpnRuntimeState.updateNetworkWarning(null)
             DiagnosticLog.add("event=vpn_reconnect reason=${intent.getStringExtra(EXTRA_RECONNECT_REASON) ?: "settings_changed"}")
             startForeground(NOTIFICATION_ID, notification(this@ProxyVpnService.uiText(R.string.status_reconnecting)))
@@ -423,10 +425,12 @@ class ProxyVpnService : VpnService() {
                     failureDetail = message
                     configureHostKeyPrompt(message, promptProfileId, testOnly)
                     if (BlockingDetection.classify(message) != null) startupBlockingDetail.set(message)
-                    if (!testOnly && "dpi_hint=possible" in message) monitorHandler.post {
+                    if (!testOnly && ("dpi_hint=possible" in message || isProxyTunnelEstablished(message))) monitorHandler.post {
                         if (isStartCurrent(generation)) handleRuntimeDiagnostic(promptProfileId, message)
                     }
-                    getSystemService(NotificationManager::class.java).notify(NOTIFICATION_ID, notification(uiText(if (hostKeyPrompt != null) R.string.ssh_key_approval else if (isRunning) R.string.status_connected else R.string.status_connecting_progress)))
+                    if (!isProxyTunnelEstablished(message)) {
+                        getSystemService(NotificationManager::class.java).notify(NOTIFICATION_ID, notification(uiText(if (hostKeyPrompt != null) R.string.ssh_key_approval else if (isRunning) R.string.status_connected else R.string.status_connecting_progress)))
+                    }
                 }
             nativeStarted = started
             if (!started) {
@@ -453,8 +457,7 @@ class ProxyVpnService : VpnService() {
                 return false
             }
             underlyingNetwork?.let { setUnderlyingNetworks(arrayOf(it)) }
-            probableFailureCounts.remove(promptProfileId)
-            probableFailureTimes.remove(promptProfileId)
+            blockingEvidence.remove(promptProfileId)
             if (failoverNotice == null) VpnRuntimeState.updateNetworkWarning(null)
             else VpnRuntimeState.updateNetworkWarning(failoverNotice)
             VpnRuntimeState.updateSystem(isAlwaysOnMode, isLockdownMode, promptProfileId)
@@ -501,7 +504,7 @@ class ProxyVpnService : VpnService() {
         } else {
             val signal = BlockingDetection.classify(detail)
             if (signal != null && profileId.isNotEmpty()) {
-                val count = recordProbableFailure(profileId)
+                val count = recordProbableFailure(profileId, detail)
                 DiagnosticLog.add("event=blocking_detection result=suspected signal=${signal.name.lowercase()} consecutive=$count network=${networkKind()} profile_type=${ConfigStore(this).profile(profileId)?.config?.type?.name?.lowercase() ?: "unknown"}")
                 if (count >= 2) handleProbableBlocking(profileId, signal)
             }
@@ -592,6 +595,7 @@ class ProxyVpnService : VpnService() {
         val store = ConfigStore(this)
         val settings = store.globalConnectionSettings()
         val warning = uiText(R.string.vpn_probably_blocked, uiText(signal.titleRes))
+        blockingWarning = warning
         VpnRuntimeState.updateNetworkWarning(warning)
         // DNS failover may move a hostname away from a filtered or stale address. Give
         // the current profile one fresh resolution attempt before changing exit IP or
@@ -604,6 +608,7 @@ class ProxyVpnService : VpnService() {
                 }
             }
             val notice = this@ProxyVpnService.uiText(R.string.vpn_refresh_dns, warning)
+            blockingWarning = notice
             VpnRuntimeState.updateNetworkWarning(notice)
             DiagnosticLog.add("event=blocking_recovery action=refresh_dns profile_type=${store.profile(profileId)?.config?.type?.name?.lowercase() ?: "unknown"}")
             getSystemService(NotificationManager::class.java).notify(NOTIFICATION_ID, notification(notice))
@@ -612,6 +617,7 @@ class ProxyVpnService : VpnService() {
         }
         if (settings.failoverMode == FailoverMode.DISABLED) {
             val notice = this@ProxyVpnService.uiText(R.string.vpn_failover_disabled, warning)
+            blockingWarning = notice
             VpnRuntimeState.updateNetworkWarning(notice)
             getSystemService(NotificationManager::class.java).notify(NOTIFICATION_ID, notification(notice))
             return
@@ -628,6 +634,7 @@ class ProxyVpnService : VpnService() {
         val next = candidates.firstOrNull() ?: run {
             val notice = this@ProxyVpnService.uiText(R.string.vpn_failover_empty, warning)
             failoverNotice = notice
+            blockingWarning = notice
             store.setFailoverState(false, notice)
             VpnRuntimeState.updateNetworkWarning(notice)
             getSystemService(NotificationManager::class.java).notify(NOTIFICATION_ID, notification(notice))
@@ -648,21 +655,29 @@ class ProxyVpnService : VpnService() {
 
     private fun handleRuntimeDiagnostic(profileId: String, detail: String) {
         if (!isRunning || activeSession?.profileId != profileId) return
+        if (isProxyTunnelEstablished(detail)) {
+            blockingEvidence[profileId]?.recovered()
+            freshDnsRecoveryProfiles.remove(profileId)
+            val warning = blockingWarning ?: return
+            blockingWarning = null
+            if (VpnRuntimeState.networkWarning.value == warning) {
+                VpnRuntimeState.updateNetworkWarning(null)
+            }
+            if (failoverNotice == warning) {
+                failoverNotice = null
+                ConfigStore(this).setFailoverState(false, null)
+            }
+            DiagnosticLog.add("event=blocking_detection result=recovered")
+            return
+        }
         val signal = BlockingDetection.classify(detail) ?: return
-        val count = recordProbableFailure(profileId)
+        val count = recordProbableFailure(profileId, detail)
         if (count >= 2) handleProbableBlocking(profileId, signal)
     }
 
-    private fun recordProbableFailure(profileId: String): Int {
-        val now = SystemClock.elapsedRealtime()
-        val previous = probableFailureTimes.put(profileId, now)
-        return if (previous == null || now - previous > FAILURE_WINDOW_MS) {
-            probableFailureCounts[profileId] = 1
-            1
-        } else {
-            probableFailureCounts.merge(profileId, 1, Int::plus) ?: 1
-        }
-    }
+    private fun recordProbableFailure(profileId: String, detail: String): Int =
+        blockingEvidence.computeIfAbsent(profileId) { BlockingEvidence() }
+            .failure(SystemClock.elapsedRealtime(), detail)
 
     private fun networkKind(): String {
         val manager = getSystemService(ConnectivityManager::class.java)
@@ -844,7 +859,6 @@ class ProxyVpnService : VpnService() {
         private const val EXTRA_RECONNECT_PROFILE_ID = "reconnect_profile_id"
         private const val EXTRA_RECONNECT_REASON = "reconnect_reason"
         private const val MONITOR_INTERVAL_MS = 10_000L
-        private const val FAILURE_WINDOW_MS = 60_000L
         private const val NETWORK_CHANGE_DEBOUNCE_MS = 1_500L
         private const val MAX_MANUAL_START_FAILURES = 5
         private const val VPN_MTU = 1_400

app/src/test/java/net/megaproxy487/vpn/BlockingDetectionTest.kt

@@ -42,4 +42,32 @@ class BlockingDetectionTest {
         assertNull(BlockingDetection.classify("event=connection stage=tunnel_io reason=timeout scope=local_deadline dpi_hint=none"))
         assertEquals(BlockingSignal.TLS_HANDSHAKE_TIMEOUT, BlockingDetection.classify("event=connection stage=tls_handshake reason=timeout scope=proxy dpi_hint=possible_tls_interference"))
     }
+
+    @Test fun oneQuicSessionFailureIsNotSeveralIndependentFailures() {
+        val evidence = BlockingEvidence()
+        assertEquals(1, evidence.failure(0, "quic_session=123 conn=1"))
+        repeat(100) { assertEquals(0, evidence.failure(10, "conn=$it quic_session=123")) }
+        assertEquals(2, evidence.failure(20, "quic_session=456"))
+        assertEquals(0, evidence.failure(30, "quic_session=456"))
+    }
+
+    @Test fun failuresMustBeCloseTogetherAndUninterruptedBySuccess() {
+        val evidence = BlockingEvidence()
+        assertEquals(1, evidence.failure(0, "handshake timeout"))
+        assertEquals(1, evidence.failure(60_001, "handshake timeout"))
+        evidence.recovered()
+        assertEquals(1, evidence.failure(60_002, "quic_session=10"))
+        evidence.recovered()
+        assertEquals(1, evidence.failure(60_003, "quic_session=11"))
+        assertEquals(2, evidence.failure(60_004, "quic_session=12"))
+    }
+
+    @Test fun onlySuccessfulProxyTunnelsProveRecovery() {
+        org.junit.Assert.assertTrue(isProxyTunnelEstablished("event=connection protocol=http3 stage=tunnel result=established"))
+        org.junit.Assert.assertTrue(isProxyTunnelEstablished("event=connection protocol=http2 stage=tunnel result=established"))
+        org.junit.Assert.assertFalse(isProxyTunnelEstablished("event=connection mode=direct stage=tunnel result=established"))
+        org.junit.Assert.assertFalse(isProxyTunnelEstablished("event=connection protocol=http3 hop=jump stage=tunnel result=established"))
+        org.junit.Assert.assertFalse(isProxyTunnelEstablished("event=masque_session protocol=http3 result=established"))
+        org.junit.Assert.assertFalse(isProxyTunnelEstablished("event=connection protocol=http3 stage=tunnel result=failed"))
+    }
 }

app/src/test/java/net/megaproxy487/vpn/BlockingRecoveryTest.kt

@@ -0,0 +1,79 @@
+package net.megaproxy487.vpn
+
+import android.app.Application
+import net.megaproxy487.data.ConfigStore
+import net.megaproxy487.model.ProxyConfig
+import org.junit.After
+import org.junit.Assert.*
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.Robolectric
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.RuntimeEnvironment
+import org.robolectric.Shadows.shadowOf
+import org.robolectric.annotation.Config
+import android.os.Looper
+
+@RunWith(RobolectricTestRunner::class)
+@Config(sdk = [35], application = Application::class)
+class BlockingRecoveryTest {
+    @After fun cleanup() {
+        running(false)
+        VpnRuntimeState.updateNetworkWarning(null)
+    }
+
+    private fun running(value: Boolean) {
+        ProxyVpnService::class.java.getDeclaredField("isRunning")
+            .apply { isAccessible = true }.setBoolean(null, value)
+    }
+
+    private fun set(service: ProxyVpnService, field: String, value: Any?) {
+        ProxyVpnService::class.java.getDeclaredField(field).apply { isAccessible = true }.set(service, value)
+    }
+
+    private fun recover(service: ProxyVpnService, profile: String = "active") {
+        ProxyVpnService::class.java.getDeclaredMethod("handleRuntimeDiagnostic", String::class.java, String::class.java)
+            .apply { isAccessible = true }.invoke(service, profile,
+                "event=connection protocol=http3 stage=tunnel result=established quic_session=123")
+        shadowOf(Looper.getMainLooper()).idle()
+    }
+
+    private fun service(warning: String): ProxyVpnService {
+        val service = Robolectric.buildService(ProxyVpnService::class.java).get()
+        set(service, "activeSession", ConnectionTestTarget("active", ProxyConfig()))
+        set(service, "blockingWarning", warning)
+        running(true)
+        VpnRuntimeState.updateNetworkWarning(warning)
+        shadowOf(Looper.getMainLooper()).idle()
+        return service
+    }
+
+    @Test fun workingTunnelClearsStaleBlockingWarningAndExhaustedFailoverNotice() {
+        val warning = "Possible blocking; no alternative profiles"
+        val service = service(warning)
+        val store = ConfigStore(RuntimeEnvironment.getApplication())
+        store.setFailoverState(false, warning)
+        set(service, "failoverNotice", warning)
+        recover(service)
+        assertNull(VpnRuntimeState.networkWarning.value)
+        assertNull(store.failoverNotice())
+    }
+
+    @Test fun successPreservesOtherWarningsAndActualFailoverNotice() {
+        val service = service("Possible blocking")
+        set(service, "failoverNotice", "Using another profile")
+        VpnRuntimeState.updateNetworkWarning("Using another profile")
+        recover(service)
+        assertEquals("Using another profile", VpnRuntimeState.networkWarning.value)
+    }
+
+    @Test fun stoppedVpnAndOtherProfileCannotClearActiveWarning() {
+        val warning = "Possible blocking"
+        val service = service(warning)
+        recover(service, "obsolete")
+        assertEquals(warning, VpnRuntimeState.networkWarning.value)
+        running(false)
+        recover(service)
+        assertEquals(warning, VpnRuntimeState.networkWarning.value)
+    }
+}

docs/en/connection-metrics.md

@@ -28,7 +28,13 @@ and mixed connection-setup timings.
 The old warning based on at least 75% failed connection openings is removed.
 Existing timeout/reset-based suspected-blocking detection and recovery remain;
 its message says the proxy may be blocked **or unavailable**. It is a heuristic,
-not proof of filtering. RTT and retransmits do not trigger automatic failover.
+not proof of filtering. Recovery requires at least two independent proxy failures
+within 60 seconds without a successful proxy tunnel between them. Multiple stream
+errors from one QUIC session count once. A successful end-to-end proxy tunnel
+clears the suspicion and its warning, including an exhausted-failover warning;
+local bypass and the first jump hop alone do not prove recovery. A notice about
+an actual switch to another profile is preserved. RTT and retransmits do not
+trigger automatic failover.
 
 MASQUE displays **QUIC RTT** and **outgoing QUIC packets declared lost in the last
 five minutes** from uQUIC's passive connection statistics. RTT is the smoothed

docs/ru/connection-metrics.md

@@ -28,7 +28,13 @@
 Старое предупреждение на основе ≥75% неудачных открытий соединений удалено.
 Существующая эвристика блокировки по таймаутам/сбросам и восстановление сохранены;
 сообщение теперь допускает блокировку **или недоступность** прокси. Это не
-доказательство фильтрации. RTT и ретрансляции не запускают автоматический failover.
+доказательство фильтрации. Восстановление запускается после двух независимых сбоев
+прокси за 60 секунд без успешного прокси-туннеля между ними. Ошибки нескольких
+потоков одной QUIC-сессии учитываются один раз. Успешный туннель через всю цепочку
+сбрасывает подозрение и предупреждение, в том числе об отсутствии запасных профилей;
+локальный обход и один лишь первый Jump-узел не подтверждают восстановление.
+Уведомление о фактическом переключении на другой профиль сохраняется.
+RTT и ретрансляции не запускают автоматический failover.
 
 MASQUE показывает **QUIC RTT** и **число исходящих QUIC-пакетов, объявленных
 потерянными за последние пять минут**, из пассивной статистики uQUIC. RTT — сглаженная