Serialize Android network snapshots across monitor and core lifecycles

2026-10-27 15:04:27 +01:00
1 parent e3aa7839ae
commit 6d925fecd1
@@ -7,98 +7,134 @@ import android.os.Handler
 import android.os.Looper
 import com.utun.chat.ChatApplication

-/** ConnectivityManager передаёт актуальный конфиг сокетов в uasync без рестарта ядра.
- * Регистрация живёт в HeadlessService; состояние очереди обновлений — на main thread. */
+/**
+ * Объединяет события ConnectivityManager за 500 мс и передаёт снимок сокетов в JNI без рестарта ядра.
+ * start/stop/onCoreStarted и состояние очереди — main thread; подготовка конфига — один worker даже через stop/start.
+ * stop снимает регистрацию и запрещает публикацию; начатый IO может завершиться. SERVICE_STARTED инвалидирует прежний снимок.
+ * true от JNI означает передачу строки, не подтверждение применения: parse/reload выполняется в uasync и логируется там.
+ */
 object NetworkChangeMonitor {
-    private var started = false
-    private var generation = 0
+    private var registration: Registration? = null
+    private var coreEpoch = Any()
    private var updating = false
    private var pending = false
-    private var lastInterfaces: List<Pair<String, String>>? = null
+    private var lastInterfaces: List<Pair<String, String>>? = null // Последний снимок, переданный в JNI; не ACK применения.
    private const val DEBOUNCE_MS = 500L
    private val handler = Handler(Looper.getMainLooper())
    private val updateRunnable = Runnable { updateSockets() }

-    private val callback = object : ConnectivityManager.NetworkCallback() {
-        override fun onAvailable(network: Network) = schedule()
-        override fun onLost(network: Network) = schedule()
-        override fun onLinkPropertiesChanged(network: Network, linkProperties: LinkProperties) = schedule()
+    private class Registration(val manager: ConnectivityManager) {
+        val callback = object : ConnectivityManager.NetworkCallback() {
+            override fun onAvailable(network: Network) = schedule(this@Registration)
+            override fun onLost(network: Network) = schedule(this@Registration)
+            override fun onLinkPropertiesChanged(network: Network, linkProperties: LinkProperties) = schedule(this@Registration)
+        }
    }

-    private fun schedule() {
+    private fun schedule(owner: Registration) {
        handler.post {
-            if (!started) return@post
+            if (registration !== owner) return@post
            handler.removeCallbacks(updateRunnable)
            handler.postDelayed(updateRunnable, DEBOUNCE_MS)
        }
    }

    private fun updateSockets() {
-        if (!started || !NativeLib.isReady()) return
-        if (updating) { pending = true; return }
+        val owner = registration ?: return
+        if (!NativeLib.isReady()) {
+            LogManager.addLog("DEBUG", "NET", "socket snapshot deferred: core not ready")
+            return
+        }
+        if (updating) {
+            if (!pending) LogManager.addLog("DEBUG", "NET", "socket snapshot pending: previous config worker running")
+            pending = true
+            return
+        }
        updating = true
-        val token = generation
-        Thread {
-            var interfaces: List<Pair<String, String>>? = null
-            var config: String? = null
-            try {
-                val provider = ChatApplication.instance.configProvider
-                interfaces = provider.getAvailableInterfaces()
-                config = provider.buildConfigText()
-            } catch (e: Exception) {
-                LogManager.addLog("ERROR", "NET", "network config update failed: ${e.message}")
-            } finally {
-                val current = interfaces
-                val text = config
-                handler.post {
-                    if (!started || token != generation) return@post
-                    updating = false
-                    if (text != null && current != null && NativeLib.isReady()) {
-                        val previous = lastInterfaces?.joinToString(", ") { (n, ip) -> "$n=$ip" } ?: "n/a"
-                        val next = current.joinToString(", ") { (n, ip) -> "$n=$ip" }
-                        lastInterfaces = current
-                        LogManager.addLog("INFO", "NET", "network change: [$previous] -> [$next]")
-                        NativeLib.notifyNetworkActive()
-                        if (!NativeLib.updateSockets(text)) LogManager.addLog("ERROR", "NET", "socket update rejected")
+        pending = false
+        val token = coreEpoch
+        LogManager.addLog("DEBUG", "NET", "network config preparation started")
+        try {
+            Thread({
+                var interfaces: List<Pair<String, String>>? = null
+                var config: String? = null
+                try {
+                    val provider = ChatApplication.instance.configProvider
+                    interfaces = provider.getAvailableInterfaces()
+                    config = provider.buildConfigText()
+                } catch (e: Exception) {
+                    LogManager.addLog("ERROR", "NET", "network config update failed: ${e.javaClass.simpleName}: ${e.message}")
+                } finally {
+                    val current = interfaces
+                    val text = config
+                    handler.post {
+                        // stop не сбрасывает updating: следующий жизненный цикл ждёт этот единственный worker.
+                        updating = false
+                        val replaced = registration !== owner
+                        val restarted = coreEpoch !== token
+                        val stale = replaced || restarted
+                        if (stale) {
+                            LogManager.addLog("DEBUG", "NET", "retired network config discarded: registration=$replaced core=$restarted")
+                        } else if (text != null && current != null && NativeLib.isReady()) {
+                            val previous = lastInterfaces?.joinToString(", ") { (n, ip) -> "$n=$ip" } ?: "n/a"
+                            val next = current.joinToString(", ") { (n, ip) -> "$n=$ip" }
+                            NativeLib.notifyNetworkActive()
+                            if (NativeLib.updateSockets(text)) {
+                                lastInterfaces = current
+                                LogManager.addLog("INFO", "NET", "socket config handed to JNI: [$previous] -> [$next] chars=${text.length}")
+                            } else LogManager.addLog("ERROR", "NET", "socket update rejected by JNI")
+                        } else if (text != null && current != null) {
+                            LogManager.addLog("DEBUG", "NET", "prepared network config discarded: core not ready")
+                        }
+                        val active = registration
+                        if (active != null && (pending || stale)) {
+                            pending = false
+                            schedule(active)
+                        }
                    }
-                    if (pending) { pending = false; schedule() }
                }
-            }
-        }.start()
+            }, "utun-network-config").start()
+        } catch (e: Exception) {
+            updating = false
+            LogManager.addLog("ERROR", "NET", "network config worker launch failed: ${e.javaClass.simpleName}: ${e.message}")
+        }
    }

-    // Повторный снимок после SERVICE_STARTED покрывает изменения сети во время инициализации.
-    fun onCoreStarted() = schedule()
+    /** Повторный снимок после SERVICE_STARTED; старый worker завершится, но его конфиг в новое ядро не передаётся. */
+    fun onCoreStarted() {
+        coreEpoch = Any()
+        LogManager.addLog("DEBUG", "NET", "socket snapshot invalidated: core started")
+        registration?.let { schedule(it) }
+    }

+    /** Жизненный цикл HeadlessService: зарегистрировать ровно один callback, false — ошибка с возможностью повтора. */
    fun start(): Boolean {
-        if (started) return true
+        if (registration != null) return true
        val cm = ChatApplication.instance.getSystemService(ConnectivityManager::class.java)
        if (cm == null) { LogManager.addLog("ERROR", "NET", "ConnectivityManager unavailable"); return false }
+        val next = Registration(cm)
        try {
-            cm.registerDefaultNetworkCallback(callback)
+            cm.registerDefaultNetworkCallback(next.callback)
        } catch (e: Exception) {
-            LogManager.addLog("ERROR", "NET", "network listener registration failed: ${e.message}")
+            LogManager.addLog("ERROR", "NET", "network listener registration failed: ${e.javaClass.simpleName}: ${e.message}")
            return false
        }
-        generation++
-        started = true
+        registration = next
        LogManager.addLog("INFO", "NET", "network change listener started")
        return true
    }

+    /** Запретить публикацию/поздние callbacks; уже запущенная подготовка IO не прерывается и не дублируется новым start. */
    fun stop() {
-        if (!started) return
-        started = false
-        generation++
+        val current = registration ?: return
+        registration = null
        pending = false
-        updating = false
        lastInterfaces = null
        handler.removeCallbacks(updateRunnable)
-        val cm = ChatApplication.instance.getSystemService(ConnectivityManager::class.java)
        try {
-            cm?.unregisterNetworkCallback(callback)
+            current.manager.unregisterNetworkCallback(current.callback)
        } catch (e: Exception) {
-            LogManager.addLog("WARN", "NET", "network listener removal failed: ${e.message}")
+            LogManager.addLog("WARN", "NET", "network listener removal failed: ${e.javaClass.simpleName}: ${e.message}")
        }
        LogManager.addLog("INFO", "NET", "network change listener stopped")
    }
@@ -0,0 +1,251 @@
+package com.utun.chat.data
+
+import android.content.Context
+import android.net.ConnectivityManager
+import android.net.LinkProperties
+import android.net.Network
+import android.os.Looper
+import com.utun.chat.ChatApplication
+import java.io.IOException
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import org.junit.After
+import org.junit.Assert.*
+import org.junit.Before
+import org.junit.Test
+
+/** Настоящий monitor: debounce в виртуальной main-очереди, реальные worker и контролируемая подготовка конфига. */
+class NetworkChangeMonitorLifecycleTest {
+    private lateinit var context: Context
+    private lateinit var manager: ConnectivityManager
+    private val loop get() = Looper.getMainLooper()
+    private val provider get() = context.configProvider
+    private val blocked = mutableListOf<CountDownLatch>()
+
+    @Before fun prepare() {
+        context = Context()
+        ChatApplication.instance = context
+        manager = requireNotNull(context.connectivityManager)
+        NativeLib.ready = true
+        NativeLib.acceptSocketUpdate = true
+        NativeLib.networkNotifications.set(0)
+        NativeLib.socketUpdates.clear()
+        LogManager.entries.clear()
+    }
+
+    @After fun release() {
+        context.connectivityManager = manager
+        manager.failRemove = false
+        NetworkChangeMonitor.stop()
+        blocked.forEach { it.countDown() }
+        provider.joinWorkers()
+        loop.dispatchAll()
+        assertTrue("network callback leaked", manager.callbacks.isEmpty())
+        NativeLib.ready = true
+        NativeLib.acceptSocketUpdate = true
+    }
+
+    private fun awaitUpdates(count: Int) {
+        val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2)
+        while (NativeLib.socketUpdates.size < count && System.nanoTime() < deadline) {
+            loop.dispatchAll()
+            Thread.sleep(1)
+        }
+        assertEquals("updates=${NativeLib.socketUpdates}, logs=${LogManager.entries.takeLast(8)}", count, NativeLib.socketUpdates.size)
+    }
+
+    private fun request(index: Int = manager.callbackHistory.lastIndex) {
+        manager.callbackHistory[index].onAvailable(Network())
+        loop.dispatchAll()
+        loop.advanceBy(500)
+    }
+
+    private fun blockFirst(): Pair<CountDownLatch, CountDownLatch> {
+        val entered = CountDownLatch(1)
+        val release = CountDownLatch(1)
+        blocked.add(release)
+        provider.onBuild = { index ->
+            if (index == 1) {
+                entered.countDown()
+                check(release.await(2, TimeUnit.SECONDS)) { "test did not release first config worker" }
+            }
+            "config-$index"
+        }
+        return entered to release
+    }
+
+    @Test fun startAndStopOwnExactlyOneCallback() {
+        repeat(3) { assertTrue(NetworkChangeMonitor.start()) }
+        assertEquals(1, manager.callbacks.size)
+        repeat(3) { NetworkChangeMonitor.stop() }
+        assertTrue(manager.callbacks.isEmpty())
+    }
+
+    @Test fun burstsAreDebouncedBeforePreparingConfig() {
+        assertTrue(NetworkChangeMonitor.start())
+        manager.callbackHistory[0].onAvailable(Network())
+        loop.dispatchAll()
+        loop.advanceBy(250)
+        manager.callbackHistory[0].onLinkPropertiesChanged(Network(), LinkProperties())
+        loop.dispatchAll()
+        loop.advanceBy(250)
+        assertEquals(0, provider.builds.get())
+        loop.advanceBy(250)
+        awaitUpdates(1)
+        assertEquals(listOf("config-1"), NativeLib.socketUpdates)
+    }
+
+    @Test fun oldPlatformCallbackCannotScheduleWorkAfterRestart() {
+        val unexpected = CountDownLatch(1)
+        provider.onBuild = { unexpected.countDown(); "unexpected" }
+        assertTrue(NetworkChangeMonitor.start())
+        NetworkChangeMonitor.stop()
+        assertTrue(NetworkChangeMonitor.start())
+        request(0)
+        assertFalse("retired callback started a worker", unexpected.await(200, TimeUnit.MILLISECONDS))
+        assertEquals(0, provider.builds.get())
+        assertTrue(NativeLib.socketUpdates.isEmpty())
+    }
+
+    @Test fun stoppedWorkerCannotUpdateCore() {
+        val (entered, release) = blockFirst()
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        assertTrue(entered.await(2, TimeUnit.SECONDS))
+        NetworkChangeMonitor.stop()
+        release.countDown()
+        provider.joinWorkers()
+        loop.dispatchAll()
+        assertTrue(NativeLib.socketUpdates.isEmpty())
+        assertEquals(0, NativeLib.networkNotifications.get())
+    }
+
+    @Test fun restartWaitsForPreviousWorkerBeforeStartingNext() {
+        val (entered, release) = blockFirst()
+        val nextEntered = CountDownLatch(1)
+        val original = provider.onBuild
+        provider.onBuild = { index -> if (index == 2) nextEntered.countDown(); original(index) }
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        assertTrue(entered.await(2, TimeUnit.SECONDS))
+        NetworkChangeMonitor.stop()
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        assertFalse("overlapping config workers", nextEntered.await(200, TimeUnit.MILLISECONDS))
+        release.countDown()
+        provider.joinWorkers()
+        loop.dispatchAll()
+        assertTrue(NativeLib.socketUpdates.isEmpty())
+        loop.advanceBy(500)
+        awaitUpdates(1)
+        assertEquals(listOf("config-2"), NativeLib.socketUpdates)
+        assertEquals(1, provider.maximumConcurrentBuilds.get())
+    }
+
+    @Test fun coreStartedInvalidatesSnapshotAlreadyBeingPrepared() {
+        val (entered, release) = blockFirst()
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        assertTrue(entered.await(2, TimeUnit.SECONDS))
+        NetworkChangeMonitor.onCoreStarted()
+        release.countDown()
+        provider.joinWorkers()
+        loop.dispatchAll()
+        assertTrue("previous core snapshot was posted", NativeLib.socketUpdates.isEmpty())
+        loop.advanceBy(500)
+        awaitUpdates(1)
+        assertEquals(listOf("config-2"), NativeLib.socketUpdates)
+    }
+
+    @Test fun eventsDuringBuildProduceOneAdditionalSnapshot() {
+        val (entered, release) = blockFirst()
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        assertTrue(entered.await(2, TimeUnit.SECONDS))
+        repeat(4) { request() }
+        release.countDown()
+        provider.joinWorkers()
+        loop.dispatchAll()
+        assertEquals(listOf("config-1"), NativeLib.socketUpdates)
+        loop.advanceBy(500)
+        awaitUpdates(2)
+        assertEquals(listOf("config-1", "config-2"), NativeLib.socketUpdates)
+        assertEquals(1, provider.maximumConcurrentBuilds.get())
+    }
+
+    @Test fun coreNotReadyWaitsForStartedEvent() {
+        assertTrue(NetworkChangeMonitor.start())
+        NativeLib.ready = false
+        request()
+        assertEquals(0, provider.builds.get())
+        NativeLib.ready = true
+        NetworkChangeMonitor.onCoreStarted()
+        loop.dispatchAll()
+        loop.advanceBy(500)
+        awaitUpdates(1)
+    }
+
+    @Test fun preparationFailureIsLoggedAndNextEventCanRetry() {
+        val entered = CountDownLatch(1)
+        provider.onBuild = { index ->
+            entered.countDown()
+            if (index == 1) throw IOException("fixture config failure") else "config-$index"
+        }
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        assertTrue(entered.await(2, TimeUnit.SECONDS))
+        provider.joinWorkers()
+        loop.dispatchAll()
+        assertTrue(LogManager.entries.any { it.level == "ERROR" && it.message.contains("config") })
+        assertTrue(NativeLib.socketUpdates.isEmpty())
+        request()
+        awaitUpdates(1)
+        assertEquals(listOf("config-2"), NativeLib.socketUpdates)
+    }
+
+    @Test fun stopUsesManagerOwningRegistration() {
+        assertTrue(NetworkChangeMonitor.start())
+        context.connectivityManager = ConnectivityManager()
+        NetworkChangeMonitor.stop()
+        assertTrue(manager.callbacks.isEmpty())
+    }
+
+    @Test fun readySnapshotIsNotPostedAfterCoreStops() {
+        val (entered, release) = blockFirst()
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        assertTrue(entered.await(2, TimeUnit.SECONDS))
+        NativeLib.ready = false
+        release.countDown()
+        provider.joinWorkers()
+        loop.dispatchAll()
+        assertTrue(NativeLib.socketUpdates.isEmpty())
+        assertEquals(0, NativeLib.networkNotifications.get())
+        NativeLib.ready = true
+        NetworkChangeMonitor.onCoreStarted()
+        loop.dispatchAll()
+        loop.advanceBy(500)
+        awaitUpdates(1)
+        assertEquals(listOf("config-2"), NativeLib.socketUpdates)
+    }
+
+    @Test fun rejectedJniUpdateIsLoggedAndNextEventCanRetry() {
+        NativeLib.acceptSocketUpdate = false
+        assertTrue(NetworkChangeMonitor.start())
+        request()
+        awaitUpdates(1)
+        assertTrue(LogManager.entries.any { it.level == "ERROR" && it.message.contains("rejected") })
+        NativeLib.acceptSocketUpdate = true
+        request()
+        awaitUpdates(2)
+        assertEquals(listOf("config-1", "config-2"), NativeLib.socketUpdates)
+    }
+
+    @Test fun registrationFailureIsLoggedAndRetryable() {
+        manager.failAdd = true
+        assertFalse(NetworkChangeMonitor.start())
+        assertTrue(LogManager.entries.any { it.level == "ERROR" && it.category == "NET" })
+        manager.failAdd = false
+        assertTrue(NetworkChangeMonitor.start())
+    }
+}