fix: WebRTC group call bugs — self-event filtering, mesh setup, per-peer cleanup

Four bugs fixed in group call signaling: 1. Self-event filtering: Own answer/ICE/hangup events echoed from relays were processed as if from another peer, causing callees to attempt callee-to-callee connections with themselves. 2. Lost peer discovery: When a callee received another callee's answer while still ringing (IncomingCall state), the peer was silently discarded. Now stored in discoveredCalleePeers and used for mesh setup after accepting. 3. Duplicate onNewPeerInGroupCall: Both CallManager and CallController triggered mesh setup for the same peer, causing duplicate log entries and wasted work. Removed CallManager's redundant calls since CallController handles this internally via onCallAnswerReceived. 4. No per-peer session cleanup: When a peer hung up but the call continued, their WebRTC PeerSession was never disposed — it lingered until ICE timeout. Added onPeerLeft callback and disposePeerSession to cleanly close individual sessions and update video tracks. Also added strategic debug logging for session lifecycle, peer discovery, and disconnect handling. https://claude.ai/code/session_01SWDsQJibYi1MMgvu55puSY
2026-04-03 19:00:04 +00:00
parent 397cdb6a6b
commit bde17099a4
3 changed files with 111 additions and 17 deletions
@@ -349,6 +349,7 @@ class CallController(
        peerPubKey: HexKey,
        sdpAnswer: String,
    ) {
+        Log.d(TAG) { "onCallAnswerReceived: from=${peerPubKey.take(8)}, knownSessions=${peerSessions.keys.map { it.take(8) }}" }
        val ps = peerSessions[peerPubKey]
        if (ps != null) {
            // We have a PeerConnection for this peer — set remote description
@@ -437,10 +438,14 @@ class CallController(
     * peer with the lexicographically lower pubkey initiates.
     */
    fun onNewPeerInGroupCall(peerPubKey: HexKey) {
-        if (peerSessions.containsKey(peerPubKey)) return // already connected/connecting
+        if (peerSessions.containsKey(peerPubKey)) {
+            Log.d(TAG) { "onNewPeerInGroupCall: session already exists for ${peerPubKey.take(8)} — skipping" }
+            return
+        }

        scope.launch {
            val myPubKey = signerProvider().pubKey
+            Log.d(TAG) { "onNewPeerInGroupCall: peer=${peerPubKey.take(8)}, myPubKey=${myPubKey.take(8)}, iAmLower=${myPubKey < peerPubKey}" }
            if (myPubKey < peerPubKey) {
                Log.d(TAG) { "Initiating callee-to-callee connection to ${peerPubKey.take(8)} (I have lower pubkey)" }
                createAndOfferToPeer(peerPubKey)
@@ -671,6 +676,7 @@ class CallController(
    // ---- Per-peer PeerConnection creation ----

    private fun createPeerSession(peerPubKey: HexKey): PeerSessionState {
+        Log.d(TAG) { "createPeerSession: ${peerPubKey.take(8)}, existing sessions=${peerSessions.keys.map { it.take(8) }}" }
        val factory = peerConnectionFactory ?: throw IllegalStateException("PeerConnectionFactory not initialized")

        val session =
@@ -722,20 +728,62 @@ class CallController(
    }

    private fun onPeerDisconnected(peerPubKey: HexKey) {
-        Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected" }
+        Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected (ICE FAILED)" }
        // If all peers disconnected, hang up
+        val sessionStates = peerSessions.map { (key, ps) -> "${key.take(8)}=${ps.session.getSignalingState()}" }
+        Log.d(TAG) { "onPeerDisconnected: checking remaining sessions: $sessionStates" }
        val allDisconnected =
            peerSessions.keys.all { key ->
                key == peerPubKey || peerSessions[key]?.session?.getSignalingState() == PeerConnection.SignalingState.CLOSED
            }
        if (allDisconnected) {
+            Log.d(TAG) { "onPeerDisconnected: all peers disconnected, hanging up" }
            scope.launch { callManager.hangup() }
+        } else {
+            Log.d(TAG) { "onPeerDisconnected: other peers still active, continuing call" }
        }
    }

+    // ---- Per-peer cleanup ----
+
+    /**
+     * Disposes a single peer's WebRTC session when they leave the call
+     * but the call continues with remaining peers.
+     */
+    fun disposePeerSession(peerPubKey: HexKey) {
+        val ps = peerSessions.remove(peerPubKey)
+        if (ps != null) {
+            Log.d(TAG) { "disposePeerSession: closing session for ${peerPubKey.take(8)}, remaining sessions=${peerSessions.keys.map { it.take(8) }}" }
+            try {
+                ps.session.dispose()
+            } catch (e: Exception) {
+                Log.e(TAG, "disposePeerSession: dispose() failed for ${peerPubKey.take(8)}", e)
+            }
+            // Update remote video tracks
+            val currentTracks = _remoteVideoTracks.value
+            if (peerPubKey in currentTracks) {
+                _remoteVideoTracks.value = currentTracks - peerPubKey
+                // If the disposed peer was the primary remote track, pick a new one
+                if (_remoteVideoTrack.value == currentTracks[peerPubKey]) {
+                    stopRemoteVideoMonitor()
+                    val nextTrack = _remoteVideoTracks.value.values.firstOrNull()
+                    _remoteVideoTrack.value = nextTrack
+                    if (nextTrack != null) {
+                        startRemoteVideoMonitor(nextTrack)
+                    }
+                }
+            }
+        } else {
+            Log.d(TAG) { "disposePeerSession: no session for ${peerPubKey.take(8)} (already disposed or never created)" }
+        }
+        // Clean up any buffered ICE candidates for this peer
+        globalPendingIce.remove(peerPubKey)
+    }
+
    // ---- Cleanup ----

    fun cleanup() {
+        Log.d(TAG) { "cleanup: disposing ${peerSessions.size} peer sessions, state=${callManager.state.value::class.simpleName}" }
        // Each block is wrapped individually so that a failure in one
        // (e.g. a WebRTC native crash) does not prevent the rest from
        // running.  Without this, a single exception could leave the
@@ -225,6 +225,7 @@ class AccountViewModel(
        callManager.onIceCandidateReceived = { event -> controller.onIceCandidateReceived(event) }
        callManager.onNewPeerInGroupCall = { peerPubKey -> controller.onNewPeerInGroupCall(peerPubKey) }
        callManager.onMidCallOfferReceived = { peerPubKey, sdpOffer -> controller.onMidCallOfferReceived(peerPubKey, sdpOffer) }
+        callManager.onPeerLeft = { peerPubKey -> controller.disposePeerSession(peerPubKey) }
        callController = controller

        // Populate ActiveCallHolder so CallActivity can launch even when the app
@@ -64,10 +64,17 @@ class CallManager(
    /** Called when a mid-call offer is received from another callee in a group call. */
    var onMidCallOfferReceived: ((peerPubKey: HexKey, sdpOffer: String) -> Unit)? = null

+    /** Called when a peer leaves the call (hangup) but the call continues with remaining peers. */
+    var onPeerLeft: ((peerPubKey: HexKey) -> Unit)? = null
+
    private var timeoutJob: Job? = null
    private var resetJob: Job? = null
    private val processedEventIds = mutableSetOf<String>()

+    /** Peers whose answers we saw while still ringing (IncomingCall).
+     *  After we accept, we trigger callee-to-callee mesh setup with them. */
+    private val discoveredCalleePeers = mutableSetOf<HexKey>()
+
    companion object {
        const val CALL_TIMEOUT_MS = 60_000L // 60 seconds ringing timeout
        const val ENDED_DISPLAY_MS = 2_000L // show "call ended" briefly before resetting
@@ -220,6 +227,17 @@ class CallManager(
        val result = factory.createGroupCallAnswer(sdpAnswer, allRecipients, current.callId, signer)
        result.wraps.forEach { publishEvent(it) }
        Log.d("CallManager") { "acceptCall: answer published, now in Connecting state" }
+
+        // Trigger callee-to-callee mesh connections with peers we discovered
+        // while still ringing (their answers arrived before we accepted).
+        val discovered = discoveredCalleePeers.toSet()
+        discoveredCalleePeers.clear()
+        if (discovered.isNotEmpty()) {
+            Log.d("CallManager") { "acceptCall: triggering mesh setup with ${discovered.size} discovered peers: ${discovered.map { it.take(8) }}" }
+            for (peer in discovered) {
+                onNewPeerInGroupCall?.invoke(peer)
+            }
+        }
    }

    suspend fun rejectCall() {
@@ -243,6 +261,18 @@ class CallManager(
                "currentState=${current::class.simpleName}, sdpAnswerLength=${event.sdpAnswer().length}"
        }

+        // Self-answer: only meaningful as "answered elsewhere" in IncomingCall.
+        // In all other states it's our own echo from the relay — ignore it.
+        if (answeringPeer == signer.pubKey) {
+            if (current is CallState.IncomingCall && callId == current.callId) {
+                Log.d("CallManager") { "onCallAnswered: self-answer detected in IncomingCall — answered elsewhere" }
+                transitionToEnded(current.callId, current.peerPubKeys(), EndReason.ANSWERED_ELSEWHERE)
+            } else {
+                Log.d("CallManager") { "onCallAnswered: ignoring self-answer echo in ${current::class.simpleName}" }
+            }
+            return
+        }
+
        when (current) {
            is CallState.Offering -> {
                if (callId != current.callId) {
@@ -272,20 +302,17 @@ class CallManager(
                        )
                }
                // Forward to CallController — it routes to the correct PeerSession
-                Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding" }
+                // and internally triggers callee-to-callee mesh setup if needed.
+                Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding to CallController" }
                onAnswerReceived?.invoke(event)
-
-                // Notify callees about new peer for mesh setup
-                onNewPeerInGroupCall?.invoke(answeringPeer)
            }

            is CallState.IncomingCall -> {
                if (callId != current.callId) return
-                if (answeringPeer == signer.pubKey) {
-                    transitionToEnded(current.callId, current.peerPubKeys(), EndReason.ANSWERED_ELSEWHERE)
-                }
-                // Another group member answered — notify for future mesh connections
-                // (we'll connect to them after we accept the call ourselves)
+                // Another group member answered — remember them for mesh setup
+                // after we accept the call ourselves.
+                Log.d("CallManager") { "onCallAnswered: peer ${answeringPeer.take(8)} answered while we're still ringing, storing for later mesh" }
+                discoveredCalleePeers.add(answeringPeer)
            }

            is CallState.Connected -> {
@@ -296,13 +323,11 @@ class CallManager(
                            peerPubKeys = current.peerPubKeys + answeringPeer,
                            pendingPeerPubKeys = current.pendingPeerPubKeys - answeringPeer,
                        )
-                    // Forward to CallController for routing
-                    onAnswerReceived?.invoke(event)
-                    onNewPeerInGroupCall?.invoke(answeringPeer)
-                } else {
-                    // Renegotiation answer
-                    onAnswerReceived?.invoke(event)
                }
+                // Forward to CallController — it routes to the correct PeerSession
+                // and internally triggers callee-to-callee mesh setup if needed.
+                Log.d("CallManager") { "onCallAnswered: peer ${answeringPeer.take(8)} answer in Connected, forwarding to CallController" }
+                onAnswerReceived?.invoke(event)
            }

            else -> {
@@ -474,19 +499,24 @@ class CallManager(
        val callId = event.callId() ?: return
        val leavingPeer = event.pubKey

+        Log.d("CallManager") { "onPeerHangup: from=${leavingPeer.take(8)}, callId=$callId, state=${current::class.simpleName}" }
+
        when (current) {
            is CallState.Connected -> {
                if (callId != current.callId) return
                val connectedRemaining = current.peerPubKeys - leavingPeer
                val pendingRemaining = current.pendingPeerPubKeys - leavingPeer
                if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) {
+                    Log.d("CallManager") { "onPeerHangup: last peer left, ending call" }
                    transitionToEnded(callId, current.allPeerPubKeys, EndReason.PEER_HANGUP)
                } else {
+                    Log.d("CallManager") { "onPeerHangup: ${leavingPeer.take(8)} left, remaining=${connectedRemaining.map { it.take(8) }}, pending=${pendingRemaining.map { it.take(8) }}" }
                    _state.value =
                        current.copy(
                            peerPubKeys = connectedRemaining,
                            pendingPeerPubKeys = pendingRemaining,
                        )
+                    onPeerLeft?.invoke(leavingPeer)
                }
            }

@@ -495,13 +525,16 @@ class CallManager(
                val connectedRemaining = current.peerPubKeys - leavingPeer
                val pendingRemaining = current.pendingPeerPubKeys - leavingPeer
                if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) {
+                    Log.d("CallManager") { "onPeerHangup: last peer left during connecting, ending call" }
                    transitionToEnded(callId, current.peerPubKeys + current.pendingPeerPubKeys, EndReason.PEER_HANGUP)
                } else {
+                    Log.d("CallManager") { "onPeerHangup: ${leavingPeer.take(8)} left during connecting, remaining=${connectedRemaining.map { it.take(8) }}" }
                    _state.value =
                        current.copy(
                            peerPubKeys = connectedRemaining,
                            pendingPeerPubKeys = pendingRemaining,
                        )
+                    onPeerLeft?.invoke(leavingPeer)
                }
            }

@@ -512,6 +545,7 @@ class CallManager(
                    transitionToEnded(callId, current.peerPubKeys, EndReason.PEER_HANGUP)
                } else {
                    _state.value = current.copy(peerPubKeys = remaining)
+                    onPeerLeft?.invoke(leavingPeer)
                }
            }

@@ -542,6 +576,16 @@ class CallManager(
        }
        if (!processedEventIds.add(event.id)) return

+        // Filter out our own ICE candidates and hangups echoed back from relays.
+        // These are never useful: ICE candidates are for the remote peer, and
+        // hangups are already handled locally by hangup() → transitionToEnded.
+        // Self-answers and self-rejects are NOT filtered here because they serve
+        // as "answered/rejected elsewhere" signals when in IncomingCall state.
+        if (event.pubKey == signer.pubKey && (event is CallIceCandidateEvent || event is CallHangupEvent)) {
+            Log.d("CallManager") { "Ignoring self-event kind=${event.kind} id=${event.id.take(8)}" }
+            return
+        }
+
        Log.d("CallManager") { "Processing signaling event kind=${event.kind} id=${event.id.take(8)} state=${_state.value::class.simpleName}" }

        when (event) {
@@ -591,6 +635,7 @@ class CallManager(
        resetJob?.cancel()
        resetJob = null
        processedEventIds.clear()
+        discoveredCalleePeers.clear()
    }

    private fun transitionToEnded(