Retain failed call end commands until the worker accepts them

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

@@ -21,6 +21,7 @@ import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import java.nio.ByteBuffer
 import java.nio.ByteOrder
@@ -38,12 +39,15 @@ object CallController : TelecomCallManager.Listener {
     val state: StateFlow<CallState?> = _state
     private val mutableOutgoing = MutableStateFlow<OutgoingCallRequest?>(null)
     val outgoingRequest: StateFlow<OutgoingCallRequest?> = mutableOutgoing
-    private var mediaPreparation: Job? = null
+    @Volatile private var mediaPreparation: Job? = null
     private var radioToRestore: Long? = null
+    private val pendingEnds = MutableStateFlow<Map<Long, Boolean>>(emptyMap()) // true=decline, false=hangup; main.
+    private var endRetry: Job? = null
 
     /** Подготовка, разговор и неподтверждённый cleanup сервиса запрещают запуск чужого media.
      * Predicate читается также media workers; CallService публикует retained-владельца через volatile. */
-    fun blocksMedia(): Boolean = CallService.cleanupPending() || TelecomCallManager.cleanupPending() || mediaPreparation?.isActive == true || _state.value?.let {
+    fun blocksMedia(): Boolean = pendingEnds.value.isNotEmpty() || CallService.cleanupPending() || TelecomCallManager.cleanupPending() ||
+        mediaPreparation?.isActive == true || _state.value?.let {
         it.preparing || it.phase == CallPhase.OUTGOING || it.phase == CallPhase.RINGING ||
             it.phase == CallPhase.WAIT_DECISION || it.phase == CallPhase.ACTIVE
     } == true
@@ -156,6 +160,10 @@ object CallController : TelecomCallManager.Listener {
             LogManager.addLog("WARN", "Call", "start rejected: busy=${_state.value?.ongoing} core=$coreReady native=${NativeLib.isReady()}")
             return
         }
+        if (pendingEnds.value.isNotEmpty() || CallService.cleanupPending() || TelecomCallManager.cleanupPending()) {
+            LogManager.addLog("WARN", "Call", "start rejected: previous signalling/service cleanup pending")
+            return
+        }
         if (!mediaConfirmed && hasMedia()) {
             mutableOutgoing.value = OutgoingCallRequest(channelId, peerNodeId, peerName)
             LogManager.addLog("INFO", "Call", "outgoing requires media confirmation peer=$peerNodeId")
@@ -261,15 +269,21 @@ object CallController : TelecomCallManager.Listener {
         if (service == null || !st.ongoing || st.preparing || signallingStarted) return
         val id = st.callId
         if (st.phase == CallPhase.OUTGOING) {
+            if (!NativeLib.callStart(st.channelId, st.peerNodeId, id)) {
+                LogManager.addLog("ERROR", "Call", "start queue rejected id=$id; releasing reserved service")
+                finish(id, 7); return
+            }
             signallingStarted = true
-            NativeLib.callStart(st.channelId, st.peerNodeId, id)
             TelecomCallManager.placeCall(id, st.peerNodeId, st.peerName)
             CallRingback.startConnecting()
         } else if (st.phase == CallPhase.INCOMING && accepting) {
-            signallingStarted = true
             IncomingCallManager.stopRinging()
+            if (!NativeLib.callAccept(id)) {
+                LogManager.addLog("ERROR", "Call", "accept queue rejected id=$id; declining and releasing service")
+                end(id, 7); return
+            }
+            signallingStarted = true
             acceptSent = true
-            NativeLib.callAccept(id)
         }
     }
 
@@ -298,12 +312,39 @@ object CallController : TelecomCallManager.Listener {
             return
         }
         if (NativeLib.isReady()) {
-            if ((st.phase == CallPhase.INCOMING || st.phase == CallPhase.PENDING_INCOMING) && !acceptSent) NativeLib.callDecline(id)
-            else if (signallingStarted) NativeLib.callHangup(id)
+            if ((st.phase == CallPhase.INCOMING || st.phase == CallPhase.PENDING_INCOMING) && !acceptSent) queueEnd(id, decline = true)
+            else if (signallingStarted) queueEnd(id, decline = false)
         }
         finish(id, reason)
     }
 
+    /** Ошибка постановки не удерживает FGS/audio, но сохраняет команду и запрещает новый media до queue ACK. */
+    private fun queueEnd(id: Long, decline: Boolean) {
+        if (sendEnd(id, decline)) return
+        pendingEnds.value = pendingEnds.value + (id to decline)
+        if (endRetry?.isActive == true) return
+        endRetry = scope.launch {
+            var pause = 250L
+            while (pendingEnds.value.isNotEmpty()) {
+                delay(pause)
+                if (!coreReady || !NativeLib.isReady()) { pendingEnds.value = emptyMap(); return@launch }
+                for ((call, reject) in pendingEnds.value) if (sendEnd(call, reject)) {
+                    pendingEnds.value = pendingEnds.value - call
+                    LogManager.addLog("INFO", "Call", "Deferred ${if (reject) "decline" else "hangup"} queued id=$call")
+                }
+                pause = (pause * 2).coerceAtMost(30_000)
+            }
+        }
+    }
+    private fun sendEnd(id: Long, decline: Boolean): Boolean = try {
+        val queued = if (decline) NativeLib.callDecline(id) else NativeLib.callHangup(id)
+        if (!queued) LogManager.addLog("ERROR", "Call", "${if (decline) "decline" else "hangup"} queue rejected id=$id; retained for retry")
+        queued
+    } catch (e: Exception) {
+        LogManager.addLog("ERROR", "Call", "End command threw id=$id decline=$decline; retained for retry: $e")
+        false
+    }
+
     private fun finish(id: Long, reason: Int) {
         val st = _state.value ?: return
         if (st.callId != id || !st.ongoing) return
@@ -335,6 +376,7 @@ object CallController : TelecomCallManager.Listener {
             serviceReleased?.await()
             released?.join()
             TelecomCallManager.awaitReleased()
+            pendingEnds.first { it.isEmpty() }
             if (!coreReady || !NativeLib.isReady() || blocksMedia()) return@launch
             val group = radioToRestore ?: return@launch
             radioToRestore = null
@@ -372,6 +414,7 @@ object CallController : TelecomCallManager.Listener {
         coreReady = false
         CoreActivity.reset()
         end()
+        endRetry?.cancel(); endRetry = null; pendingEnds.value = emptyMap()
         _state.value = null
         CoreActivity.call(false)
     }
@@ -454,7 +497,7 @@ object CallController : TelecomCallManager.Listener {
             if (current?.callId == id) return
             if (current?.ongoing == true) {
                 LogManager.addLog("WARN", "Call", "busy; decline incoming id=$id current=${current.callId}")
-                NativeLib.callDecline(id)
+                queueEnd(id, decline = true)
                 return
             }
             incomingAt = SystemClock.elapsedRealtime()
@@ -474,6 +517,7 @@ object CallController : TelecomCallManager.Listener {
             (type == 39 && data.size != 24) ||
             (type == 44 && (data.size != 17 || (data[8].toInt() and 0xFF) > 3))) { badEvent(type, data); return }
         val id = ByteBuffer.wrap(data, 0, 8).order(ByteOrder.LITTLE_ENDIAN).long
+        if (type in listOf(37, 38, 40)) pendingEnds.value = pendingEnds.value - id
         val st = _state.value
         if (st == null || st.callId != id || !st.ongoing) {
             LogManager.addLog("DEBUG", "Call", "ignore event=$type id=$id current=${st?.callId} phase=${st?.phase}")
@@ -591,8 +635,11 @@ object CallController : TelecomCallManager.Listener {
         val st = _state.value ?: return
         if (st.callId != id || st.phase != CallPhase.WAIT_DECISION) return
         LogManager.addLog("INFO", "Call", "Caller overrides phone busy id=$id")
+        if (!NativeLib.callContinue(id)) {
+            LogManager.addLog("ERROR", "Call", "continue queue rejected id=$id; caller choice remains available")
+            return
+        }
         _state.value = st.copy(phase = CallPhase.OUTGOING)
-        NativeLib.callContinue(id)
         CallRingback.startConnecting()
         service?.updateNotification()
     }

tools/chatgui-android/router-tests/src/main/kotlin/com/utun/chat/data/RouterHostEnvironment.kt

@@ -54,11 +54,12 @@ object NativeLib {
     fun callAudioSetAec(enabled: Boolean) {}
     fun callAudioSetAecDelay(frames: Int) {}
     val callCommands = mutableListOf<Pair<String, Long>>()
-    fun callStart(channel: String, peer: Long, id: Long) { callCommands.add("start" to id) }
-    fun callAccept(id: Long) { callCommands.add("accept" to id) }
-    fun callDecline(id: Long) { callCommands.add("decline" to id) }
-    fun callHangup(id: Long) { callCommands.add("hangup" to id) }
-    fun callContinue(id: Long) { callCommands.add("continue" to id) }
+    var rejectedCallCommands = emptySet<String>()
+    fun callStart(channel: String, peer: Long, id: Long): Boolean { callCommands.add("start" to id); return "start" !in rejectedCallCommands }
+    fun callAccept(id: Long): Boolean { callCommands.add("accept" to id); return "accept" !in rejectedCallCommands }
+    fun callDecline(id: Long): Boolean { callCommands.add("decline" to id); return "decline" !in rejectedCallCommands }
+    fun callHangup(id: Long): Boolean { callCommands.add("hangup" to id); return "hangup" !in rejectedCallCommands }
+    fun callContinue(id: Long): Boolean { callCommands.add("continue" to id); return "continue" !in rejectedCallCommands }
     data class RadioCallCommand(val id: Long, val action: Int, val group: Long, val peer: Long)
     val radioCalls = mutableListOf<RadioCallCommand>()
     var onRadioCall: (RadioCallCommand) -> Boolean = { true }