Keep publisher service owners until capture and foreground cleanup are acknowledged

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

@@ -73,6 +73,7 @@ class StreamCaptureEngine(
     private var videoReleasing = false
     private var videoReleaseSince = 0L
     private var slowReleaseReported = false
+    private var stopVideoRequested = false
     private var generation = 0L
     private var videoRevision = 0L
     private var videoRotation = 0
@@ -502,7 +503,7 @@ class StreamCaptureEngine(
     /** Только capture worker: независимый cleanup, refs снимаются после возврата SDK либо callback закрытия. */
     private fun cleanupResources() {
         if (stopping) {
-            if (!videoReleasing) releaseVideo {}
+            if (!stopVideoRequested) { stopVideoRequested = true; releaseVideo {} }
             releaseDone = null; audioRunning = false
             try { observeOrientation(false) } catch (e: Exception) {
                 LogManager.addLog("ERROR", "video", "capture orientation release failed id=$id; retained: $e")
@@ -542,7 +543,9 @@ class StreamCaptureEngine(
                 if (!stopping) try { done?.invoke() } catch (e: Exception) { failed("capture continuation: $e") }
             }
         }
-        if (stopping && !videoReleasing && cameraClosed && audioCapture == null && audioThread == null && !orientationEnabled) {
+        if (stopping && !videoReleasing && !opening && camera == null && closingCameras.isEmpty() &&
+            ownedSessions.isEmpty() && cameraSessions.isEmpty() && encoderResources == null && input == null &&
+            audioCapture == null && audioThread == null && !orientationEnabled) {
             stopped = true; retained.remove(this)
             synchronized(this) { handler.removeCallbacks(cleanupTask); cleanupQueued = false }
             thread.quitSafely()

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

@@ -11,6 +11,8 @@ import android.os.IBinder
 import android.os.Handler
 import android.os.Looper
 import android.os.PowerManager
+import android.os.SystemClock
+import android.net.Uri
 import androidx.core.app.NotificationCompat
 import androidx.core.app.ServiceCompat
 import com.utun.chat.MainActivity
@@ -27,9 +29,14 @@ import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.collect
+import java.util.concurrent.ConcurrentHashMap
 
-/** Foreground-владелец камеры, микрофона и CPU lock; не восстанавливает live session после смерти процесса. */
+/** Main-владелец capture/native publisher/CPU lock/notification. Сохраняет refs до ACK, также после onDestroy.
+ * Handler повторяет независимые ошибки; FIN следует после capture.stop=true, доставка ждёт до 4 секунд.
+ * Live session не восстанавливается после смерти процесса.
+ */
 class StreamService : Service() {
+    companion object { private val retained = ConcurrentHashMap.newKeySet<StreamService>() }
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate)
     private var capture: StreamCaptureEngine? = null
     private var lock: PowerManager.WakeLock? = null
@@ -37,27 +44,50 @@ class StreamService : Service() {
     private var stopping = false
     private var aborting = false
     private var poll: Job? = null
+    private val main = Handler(Looper.getMainLooper())
+    private val retry = Runnable { cleanup() }
+    private var retryDelay = 250L
+    private var nativeStopped = false
+    private var drainUntil = 0L
+    private var destroyed = false
+    private var foregroundOwned = false
+    private var stopRequested = false
+    private var controllerEnded = false
+    private var lastStartId = 0
+    private var openIntent: PendingIntent? = null
+    private var stopIntent: PendingIntent? = null
 
     override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
-        if (intent?.action == "stop") { StreamController.stop(); return START_NOT_STICKY }
-        if (id != 0L) return START_NOT_STICKY
+        lastStartId = startId
+        if (intent?.action == "stop") {
+            val requested = intent.getLongExtra("id", 0)
+            if (requested != 0L && requested == StreamController.state.value?.id) StreamController.stop()
+            else LogManager.addLog("WARN", "stream", "stale notification stop ignored requested=$requested current=${StreamController.state.value?.id}")
+            if (id == 0L) finish()
+            return START_NOT_STICKY
+        }
+        if (id != 0L || stopping) {
+            LogManager.addLog("WARN", "stream", "duplicate service start ignored id=$id stopping=$stopping")
+            return START_NOT_STICKY
+        }
         id = intent?.getLongExtra("id", 0) ?: 0
         val audio = intent?.getBooleanExtra("audio", true) ?: true
         try {
             val notifications = getSystemService(NotificationManager::class.java)
             notifications.createNotificationChannel(NotificationChannel("live_stream", getString(R.string.stream_title), NotificationManager.IMPORTANCE_LOW))
-            val open = PendingIntent.getActivity(this, 0, Intent(this, MainActivity::class.java), PendingIntent.FLAG_IMMUTABLE)
-            val stop = PendingIntent.getService(this, 0, Intent(this, StreamService::class.java).setAction("stop"), PendingIntent.FLAG_IMMUTABLE)
+            val open = PendingIntent.getActivity(this, 0, Intent(this, MainActivity::class.java).setData(Uri.parse("utun-stream:open:$id")),
+                PendingIntent.FLAG_IMMUTABLE).also { openIntent = it }
+            val stop = PendingIntent.getService(this, 0, Intent(this, StreamService::class.java).setAction("stop")
+                .setData(Uri.parse("utun-stream:stop:$id")).putExtra("id", id), PendingIntent.FLAG_IMMUTABLE).also { stopIntent = it }
             val notification = NotificationCompat.Builder(this, "live_stream").setSmallIcon(android.R.drawable.ic_menu_camera)
                 .setContentTitle(getString(R.string.stream_live)).setContentIntent(open).setOngoing(true)
                 .addAction(android.R.drawable.ic_media_pause, getString(R.string.stream_stop), stop).build()
             val type = if (Build.VERSION.SDK_INT >= 30) ServiceInfo.FOREGROUND_SERVICE_TYPE_CAMERA or
                 (if (audio) ServiceInfo.FOREGROUND_SERVICE_TYPE_MICROPHONE else 0) else 0
-            ServiceCompat.startForeground(this, 1004, notification, type)
+            foregroundOwned = true; ServiceCompat.startForeground(this, 1004, notification, type)
             if (id == 0L || !StreamController.serviceReady(id, this)) { finish(); return START_NOT_STICKY }
-            lock = getSystemService(PowerManager::class.java).newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "utun:live_stream").apply {
-                setReferenceCounted(false); acquire()
-            }
+            lock = getSystemService(PowerManager::class.java).newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "utun:live_stream")
+            lock!!.setReferenceCounted(false); lock!!.acquire()
             val group = intent!!.getLongExtra("group", 0)
             val publisher = StreamController.state.value ?: error("Missing publisher")
             val config = publisher.pendingConfig ?: error("Missing capture configuration")
@@ -65,7 +95,8 @@ class StreamService : Service() {
             val audioSettings = if (audio) runBlocking { provider.radioCompressorSettings.first() } else RadioCompressorSettings()
             capture = StreamCaptureEngine(this, id, group, config, audio, publisher.preview, audioSettings,
                 { active, restored -> StreamController.captureReady(id, active, restored) },
-                { detail -> StreamController.failed(id, detail) }).also { it.start() }
+                { detail -> StreamController.failed(id, detail) })
+            capture!!.start()
             if (audio) scope.launch {
                 try { provider.radioCompressorSettings.collect { capture?.audioCompressorSettings = it } }
                 catch (e: CancellationException) { throw e }
@@ -74,6 +105,7 @@ class StreamService : Service() {
                 }
             }
             poll = scope.launch {
+                try {
                 val deadline = System.nanoTime() + 10_000_000_000
                 var opened = false
                 var nextStats = 0L
@@ -94,6 +126,11 @@ class StreamService : Service() {
                     }
                     delay(20)
                 }
+                } catch (e: CancellationException) { throw e }
+                catch (e: Exception) {
+                    LogManager.addLog("ERROR", "stream", "publisher polling failed id=$id: $e")
+                    StreamController.failed(id, "publisher polling: $e"); shutdown(abort = true)
+                }
             }
             LogManager.addLog("INFO", "stream", "foreground started id=$id group=$group microphone=$audio CPU lock=${lock?.isHeld}")
         } catch (e: Exception) {
@@ -117,42 +154,64 @@ class StreamService : Service() {
 
     /** Producers завершатся до FIN; при уничтожении ядра очередь не ждём. */
     fun shutdown(abort: Boolean = false) {
+        check(Looper.myLooper() == Looper.getMainLooper()) { "Stream service shutdown requires main" }
         aborting = aborting || abort
-        if (stopping) return
-        stopping = true
-        val engine = capture
-        if (engine == null) { NativeLib.streamStop(id); finish(); return }
-        engine.stop {
-            scope.launch {
-                capture = null; NativeLib.streamStop(id)
-                if (aborting) finish()
-                else scope.launch { delay(4000); if (stopping) finish() }
-            }
-        }
+        stopping = true; retained.add(this); main.removeCallbacks(retry); cleanup()
     }
 
     private fun finish() {
-        if (capture != null) { shutdown(abort = true); return }
-        poll?.cancel(); poll = null
-        lock?.let { if (it.isHeld) it.release() }; lock = null
-        StreamController.ended(id, this)
-        stopForeground(STOP_FOREGROUND_REMOVE); stopSelf()
+        shutdown(abort = true)
+    }
+
+    /** Main остаётся доступен для SDK callbacks; false от engine сохраняет владельца до следующей попытки. */
+    private fun cleanup() {
+        main.removeCallbacks(retry)
+        capture?.let { owned -> try {
+            if (owned.stop()) capture = null
+            else LogManager.addLog("DEBUG", "stream", "publisher capture cleanup pending id=$id")
+        } catch (e: Exception) { LogManager.addLog("ERROR", "stream", "publisher capture stop failed id=$id; retained: $e") } }
+        if (capture == null && !nativeStopped) try {
+            if (id != 0L) NativeLib.streamStop(id)
+            nativeStopped = true; drainUntil = SystemClock.elapsedRealtime() + 4000
+            LogManager.addLog("INFO", "stream", "publisher producers released id=$id; FIN requested abort=$aborting")
+        } catch (e: Exception) { LogManager.addLog("ERROR", "stream", "publisher native stop failed id=$id; retained: $e") }
+        val draining = !aborting && !destroyed && nativeStopped && SystemClock.elapsedRealtime() < drainUntil
+        if (capture == null && nativeStopped && !draining) {
+            poll?.cancel(); poll = null
+            lock?.let { owned -> try { if (owned.isHeld) owned.release(); lock = null } catch (e: Exception) {
+                LogManager.addLog("ERROR", "stream", "publisher CPU lock release failed id=$id; retained: $e")
+            } }
+            openIntent?.let { owned -> try { owned.cancel(); openIntent = null } catch (e: Exception) {
+                LogManager.addLog("ERROR", "stream", "publisher open token cancellation failed id=$id; retained: $e")
+            } }
+            stopIntent?.let { owned -> try { owned.cancel(); stopIntent = null } catch (e: Exception) {
+                LogManager.addLog("ERROR", "stream", "publisher stop token cancellation failed id=$id; retained: $e")
+            } }
+            if (foregroundOwned) try { stopForeground(STOP_FOREGROUND_REMOVE); foregroundOwned = false } catch (e: Exception) {
+                LogManager.addLog("ERROR", "stream", "publisher foreground removal failed id=$id; retained: $e")
+            }
+            if (!stopRequested) try { if (!destroyed) stopSelf(lastStartId); stopRequested = true } catch (e: Exception) {
+                LogManager.addLog("ERROR", "stream", "publisher service stop failed id=$id; retained: $e")
+            }
+            if (lock == null && openIntent == null && stopIntent == null && !foregroundOwned && stopRequested && !controllerEnded) try {
+                StreamController.ended(id, this); controllerEnded = true
+            } catch (e: Exception) { LogManager.addLog("ERROR", "stream", "publisher completion failed id=$id; retained: $e") }
+        }
+        if (controllerEnded) {
+            retained.remove(this); retryDelay = 250
+            LogManager.addLog("INFO", "stream", "publisher service cleanup confirmed id=$id")
+        } else {
+            val delay = if (draining) minOf(250, drainUntil - SystemClock.elapsedRealtime()).coerceAtLeast(1) else retryDelay
+            if (!draining) retryDelay = (retryDelay * 2).coerceAtMost(30_000)
+            if (!main.postDelayed(retry, delay)) LogManager.addLog("ERROR", "stream", "publisher cleanup retry rejected id=$id; owner retained")
+            else LogManager.addLog("DEBUG", "stream", "publisher cleanup scheduled id=$id delayMs=$delay capture=${capture != null} drain=$draining")
+        }
     }
 
     override fun onDestroy() {
-        poll?.cancel()
-        val engine = capture; capture = null
-        val ownedLock = lock; lock = null
-        val release = {
-            NativeLib.streamStop(id)
-            ownedLock?.let { if (it.isHeld) it.release() }
-            Handler(Looper.getMainLooper()).post { StreamController.ended(id, this) }
-            Unit
-        }
-        if (engine != null) engine.stop(release) else release()
-        scope.cancel()
-        LogManager.addLog("INFO", "stream", "foreground destroyed id=$id")
-        super.onDestroy()
+        destroyed = true; scope.cancel()
+        try { shutdown(abort = true); LogManager.addLog("INFO", "stream", "foreground destroyed id=$id pending=${!controllerEnded}") }
+        finally { super.onDestroy() }
     }
     override fun onBind(intent: Intent?): IBinder? = null
 }