From 6d925fecd1a774a14bbbfe0bcf4d9c6bfcd2ff2c Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 15:02:55 +0100 Subject: [PATCH] Serialize Android network snapshots across monitor and core lifecycles --- .../utun/chat/data/NetworkChangeMonitor.kt | 142 ++++++---- .../data/NetworkChangeMonitorLifecycleTest.kt | 251 ++++++++++++++++++ 2 files changed, 340 insertions(+), 53 deletions(-) create mode 100644 tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/NetworkChangeMonitorLifecycleTest.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NetworkChangeMonitor.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NetworkChangeMonitor.kt index 5834d93a..f3bffd60 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NetworkChangeMonitor.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NetworkChangeMonitor.kt @@ -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>? = null + private var lastInterfaces: List>? = 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>? = 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>? = 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") } diff --git a/tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/NetworkChangeMonitorLifecycleTest.kt b/tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/NetworkChangeMonitorLifecycleTest.kt new file mode 100644 index 00000000..78c5ab1c --- /dev/null +++ b/tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/NetworkChangeMonitorLifecycleTest.kt @@ -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() + + @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 { + 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()) + } +}