Serialize native lifecycle mutations and renew readiness across restart

tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt

@@ -3,6 +3,8 @@ package com.utun.chat.data
 import com.utun.chat.ChatApplication
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 
 interface ChatEventCallback {
     fun onEvent(type: Int, data: ByteArray?)
@@ -11,15 +13,16 @@ interface ChatEventCallback {
 object NativeLib {
     @Volatile private var coreStopping = false
     @Volatile private var eventCallback: ChatEventCallback? = null
+    private val lifecycle = Mutex()
+    private var initialized = false // Только под lifecycle; callbacks нельзя менять у ещё живого worker.
     fun isCoreStopping(): Boolean = coreStopping
 
     init {
         System.loadLibrary("utun")
     }
 
-    /** Save Kotlin callbacks to C globals (log, config). Must be called before start(). */
-    fun init(dbPath: String, controlPort: Int): Boolean {
-        if (coreStopping) { LogManager.addLog("ERROR", "Core", "Initialization rejected: frontend cleanup pending"); return false }
+    /** Только IO под lifecycle, до первого worker либо после подтверждённого nativeStop. */
+    private fun initCallbacks(dbPath: String, controlPort: Int): Boolean {
         val provider = ChatApplication.instance.configProvider
         val settings = provider.getVoiceSettings()
         val ok = nativeInit(
@@ -37,20 +40,22 @@ object NativeLib {
         nativeVoiceSetCompressorConfig(settings.maxGainDb, settings.lookbackMs, settings.lookaheadMs,
             settings.riseRate.toFloat(), settings.targetLevelDb.toFloat())
         LogManager.addLog("INFO", "AudioSettings", "Applied at initialization: $settings")
+        initialized = true
         return true
     }
 
-    /** Start full uTun instance from INI config text */
-    fun start(configText: String): Boolean {
-        if (coreStopping) { LogManager.addLog("ERROR", "Core", "Start rejected: previous frontend cleanup pending"); return false }
-        if (isReady()) return true
-        return nativeStart(configText)
+    /** Инициализация callbacks и создание worker — одна операция; существующий worker сохраняет callbacks до join. */
+    suspend fun start(dbPath: String, controlPort: Int, configText: String): Boolean = lifecycle.withLock {
+        if (coreStopping) { LogManager.addLog("ERROR", "Core", "Start rejected: previous frontend cleanup pending"); return@withLock false }
+        if (isReady()) return@withLock true
+        withContext(Dispatchers.IO) { (initialized || initCallbacks(dbPath, controlPort)) && nativeStart(configText) }
     }
 
     /** Снимает frontend media на main, дожидается worker на IO даже до ready; ошибка запрещает следующий init. */
-    suspend fun stop() {
+    suspend fun stop() = lifecycle.withLock {
         prepareForCoreStop()
         check(withContext(Dispatchers.IO) { nativeStop() }) { "native worker stop failed" }
+        initialized = false
         withContext(Dispatchers.Main.immediate) { ChatApplication.instance.memberFiles.coreStopped() }
         coreStopping = false
     }
@@ -61,16 +66,43 @@ object NativeLib {
     }
 
     /** Как stop, дополнительно освобождает JNI global refs только после подтверждённого join. */
-    suspend fun destroy() {
+    suspend fun destroy() = lifecycle.withLock {
         prepareForCoreStop()
         check(withContext(Dispatchers.IO) { nativeDestroy() }) { "native worker destroy failed" }
+        initialized = false
         withContext(Dispatchers.Main.immediate) { ChatApplication.instance.memberFiles.coreStopped() }
         coreStopping = false
     }
 
-    suspend fun restart(dbPath: String, controlPort: Int, configText: String): Boolean {
-        try { prepareForCoreStop(); return withContext(Dispatchers.IO) { nativeRestart(configText) } }
-        finally { coreStopping = false }
+    suspend fun restart(dbPath: String, controlPort: Int, configText: String): Boolean = lifecycle.withLock {
+        var accepted = false
+        try {
+            prepareForCoreStop()
+            withContext(Dispatchers.Main.immediate) { com.utun.chat.headless.HeadlessService.onCoreRestarting() }
+            accepted = withContext(Dispatchers.IO) { (initialized || initCallbacks(dbPath, controlPort)) && nativeRestart(configText) }
+            accepted
+        } finally { resumeAfterRejectedChange(accepted) }
+    }
+
+    /** Тот же mutex исключает start/stop между join, удалением БД и восстановлением worker.
+     * afterStop выполняется на main после join; configText=null оставляет ядро выключенным. */
+    suspend fun resetDatabase(dbPath: String, controlPort: Int, configText: String?, afterStop: () -> Unit): Boolean = lifecycle.withLock {
+        prepareForCoreStop()
+        if (configText != null) withContext(Dispatchers.Main.immediate) { com.utun.chat.headless.HeadlessService.onCoreRestarting() }
+        check(withContext(Dispatchers.IO) { nativeStop() }) { "native stop before database reset failed" }
+        initialized = false
+        withContext(Dispatchers.Main.immediate) { ChatApplication.instance.memberFiles.coreStopped(); afterStop() }
+        val cleared = withContext(Dispatchers.IO) { nativeClearDatabase() }
+        if (configText != null) check(withContext(Dispatchers.IO) {
+            initCallbacks(dbPath, controlPort) && nativeStart(configText)
+        }) { "native start after database reset failed" }
+        coreStopping = false
+        cleared
+    }
+
+    private fun resumeAfterRejectedChange(accepted: Boolean) {
+        coreStopping = false
+        if (!accepted && nativeIsRunning()) com.utun.chat.AppEventHandler.onEvent(12, null)
     }
 
     /** Снять frontend-владельцев перед stop/restart/сменой ключей. Неподтверждённый CallService либо ещё
@@ -84,6 +116,7 @@ object NativeLib {
             StreamController.reset(); CallController.reset(); ChatApplication.instance.avatars.stop(); ChatApplication.instance.memberFiles.cancel()
             radio
         }
+        LogManager.addLog("INFO", "Core", "Waiting for radio audio cleanup")
         withContext(Dispatchers.IO) { radioStopped.get() }
         withContext(Dispatchers.Main.immediate) {
             CallService.requireReleased() // Не останавливать native при неподтверждённой очистке его media-владельца.
@@ -95,6 +128,7 @@ object NativeLib {
                 "Call media preparation still running"
             }
         }
+        LogManager.addLog("INFO", "Core", "Waiting for shared media audio cleanup")
         withContext(Dispatchers.IO) { MediaAudioController.awaitIdle() }
     }
 
@@ -162,9 +196,14 @@ object NativeLib {
     fun subscribeNativeLogs(enabled: Boolean) { nativeSubscribeLogs(enabled) }
     fun setUdpLogTarget(ip: String, port: Int) { nativeSetUdpLogTarget(ip, port) }
     /** Принимает полный INI для пересоздания ядра с новой сохранённой идентичностью; true подтверждает приём запроса. */
-    suspend fun regenerateKeys(configText: String): Boolean {
-        try { prepareForCoreStop(); return withContext(Dispatchers.IO) { nativeRegenerateKeys(configText) } }
-        finally { coreStopping = false }
+    suspend fun regenerateKeys(configText: String): Boolean = lifecycle.withLock {
+        var accepted = false
+        try {
+            prepareForCoreStop()
+            withContext(Dispatchers.Main.immediate) { com.utun.chat.headless.HeadlessService.onCoreRestarting() }
+            accepted = withContext(Dispatchers.IO) { nativeRegenerateKeys(configText) }
+            accepted
+        } finally { resumeAfterRejectedChange(accepted) }
     }
 
     /* ── Channel members ── */

tools/chatgui-android/app/src/main/java/com/utun/chat/headless/CoreStartup.kt

@@ -20,13 +20,14 @@ internal class CoreStartup(
     private val log: (String, String) -> Unit,
     private val timeoutMillis: Long = 10_000,
 ) {
-    private val ready = CompletableDeferred<Unit>()
+    private var ready = CompletableDeferred<Unit>()
     private var job: Job? = null
+    private var recovery: Job? = null
     private var requested = false
     private var closed = false
 
     /** prepare инициализирует frontend/сеть и возвращает INI; submit=true подтверждает только создание native-потока. */
-    fun start(prepare: suspend () -> String, submit: (String) -> Boolean) {
+    fun start(prepare: suspend () -> String, submit: suspend (String) -> Boolean) {
         if (job != null || closed) {
             log("WARN", "startup ignored: requested=$requested closed=$closed")
             return
@@ -54,6 +55,7 @@ internal class CoreStartup(
 
     /** Успешный await подтверждает готовность ядра и завершённый onReady; ошибка запуска распространяется всем ожидающим. */
     suspend fun awaitReady() {
+        if (!closed && requested && ready.isCompleted && !isReady()) restarting()
         ready.await()
         check(!closed) { "service startup is closed" }
     }
@@ -65,14 +67,35 @@ internal class CoreStartup(
         }
         try {
             onReady()
+            recovery?.cancel(); recovery = null
             if (ready.complete(Unit)) log("INFO", "initial core startup completed")
         } catch (e: Exception) { fail(e) }
     }
 
+    /** Закрыть прежнюю готовность до async restart; тот же deferred ждёт новый/возобновлённый ready. */
+    fun restarting() {
+        if (closed || !requested) return
+        if (ready.isCompleted) ready = CompletableDeferred()
+        recovery?.cancel()
+        val pending = ready
+        recovery = scope.launch {
+            try { withTimeout(timeoutMillis) { pending.await() } }
+            catch (e: TimeoutCancellationException) { fail(IllegalStateException("core recovery timeout after ${timeoutMillis}ms", e)) }
+            catch (e: CancellationException) { throw e }
+            catch (e: Exception) { if (!closed) fail(e) }
+        }
+        log("INFO", "core readiness suspended; waiting for restart/recovery")
+    }
+
+    fun coreStopped() {
+        if (!closed && requested && ready.isCompleted) restarting()
+    }
+
     /** Также используется для ошибок init/регистрации сети до start; stopSelf вызывается один раз. */
     fun fail(error: Exception) {
         if (closed) return
         closed = true
+        recovery?.cancel(); recovery = null
         ready.completeExceptionally(error)
         log("ERROR", "core startup failed: ${error.javaClass.simpleName}: ${error.message}")
         try { onFailure() }
@@ -87,5 +110,6 @@ internal class CoreStartup(
             log("INFO", "service startup closed")
         }
         job?.cancel()
+        recovery?.cancel(); recovery = null
     }
 }

tools/chatgui-android/app/src/main/java/com/utun/chat/headless/HeadlessService.kt

@@ -49,6 +49,8 @@ class HeadlessService : Service() {
         }
 
         fun onCoreStarted() { currentService?.startup?.coreStarted() }
+        fun onCoreStopped() { currentService?.startup?.coreStopped() }
+        fun onCoreRestarting() { currentService?.startup?.restarting() }
 
         fun refreshLanguage() {
             currentService?.let { service ->
@@ -82,6 +84,7 @@ class HeadlessService : Service() {
             return
         }
 
+        var startupDirectory = filesDir.absolutePath
         startup.start(prepare = {
             shutdown.await()
             val provider = ChatApplication.instance.configProvider
@@ -89,10 +92,10 @@ class HeadlessService : Service() {
                 (provider.getString("db.path") ?: filesDir.absolutePath) to provider.getInt("node.listen_port")
             }
             LogManager.addLog("INFO", "SRV", "uTun service starting, port=$listenPort db=$dbDir")
-            check(NativeLib.init(dbDir, 9999)) { "native initialization failed" }
+            startupDirectory = dbDir
             check(NetworkChangeMonitor.start()) { "network monitoring unavailable" }
             provider.buildConfigText().await()
-        }, submit = NativeLib::start)
+        }, submit = { config -> NativeLib.start(startupDirectory, 9999, config) })
     }
 
     override fun onDestroy() {