feat(call): per-peer 30s invite timeout and per-peer status in video grid

Two changes that tighten how the caller-side UX handles slow or unresponsive callees in a group call: 1. Per-peer 30-second invite timeout Previously the caller used a single 60-second call-wide timeout: if *any* peer answered within the window the timer was cancelled and slow peers could remain "ringing" indefinitely. For a mid-call invite there was no timeout at all — the invitee stayed in pendingPeerPubKeys forever, burning a PeerConnection on the caller side and keeping the invitee's device ringing. CallManager now schedules an independent 30-second timer for every peer in pendingPeerPubKeys (or Offering.peerPubKeys in the initial offer phase). The timer is started when the peer is added to pending — in beginOffering, initiateCall and invitePeer — and is cancelled when the peer answers (onCallAnswered), rejects (onCallRejected) or hangs up (onPeerHangup). On expiry the peer is dropped from the group, a CallHangup is published to them so their device stops ringing, and onPeerLeft fires so CallController can dispose the per-peer PeerConnection. If the drop leaves the caller with zero connected and zero pending peers, the call ends with EndReason.TIMEOUT. Callee ringing still uses the 60-second CALL_TIMEOUT_MS in onIncomingCallEvent; the two timers now serve distinct roles. New constant CallManager.PEER_INVITE_TIMEOUT_MS = 30_000L. NIP-AC.md "Event Lifecycle" documents both timers. 2. Per-peer "Calling..." status in the video grid The shared "Waiting for others to join…" banner across the top of ConnectedCallUI is removed. PeerVideoGrid now takes a pendingPeerPubKeys set and, for each peer still pending, routes rendering through PeerAvatarCell with a "Calling…" status line under the username — so it's obvious *which* participants the call is waiting on, not just *that* it's waiting. A peer in pending never shows as a video cell even if a stale track is still in the map, because they haven't actually answered yet. The voice-only path in ConnectedCallUI now also uses PeerVideoGrid (with empty tracks/active-video) instead of the single GroupCallPictures + GroupCallNames stack, so the per-peer status behavior is consistent for audio calls. Tests added in CallManagerTest: - perPeerTimeoutEndsP2PCallWhenBobNeverAnswers - perPeerTimeoutDropsSlowCalleeFromGroupCall - perPeerTimeoutIsCancelledOnAnswer - perPeerTimeoutDropsMidCallInviteeWhenNoAnswer - perPeerTimeoutIsCancelledOnReject - perPeerTimeoutEndsCallWhenAllCalleesIgnore All existing CallManagerTest cases still pass unchanged. https://claude.ai/code/session_01XSPDbahLwHs9sdF5XfRvHB
2026-04-14 23:23:42 +00:00
parent d087e1ac13
commit b9d6cefbeb
5 changed files with 362 additions and 39 deletions
@@ -79,6 +79,14 @@ class CallManager(
    private var resetJob: Job? = null
    private val processedEventIds = LinkedHashSet<String>()

+    /** Per-peer invite timeout jobs.  A separate 30-second timer is scheduled
+     *  for each peer we are waiting on (initial group-call offerees and
+     *  mid-call invitees) so that slow/unavailable peers can be dropped
+     *  individually without affecting the rest of the call. The timer is
+     *  cancelled when the peer answers, rejects, or hangs up — or when the
+     *  whole call ends. */
+    private val perPeerTimeoutJobs = mutableMapOf<HexKey, Job>()
+
    /** Call IDs for which we have seen a hangup, reject, or answer-elsewhere
     *  signal.  Checked before transitioning to [CallState.IncomingCall] so
     *  that stale offer events replayed by relays after an app restart do not
@@ -95,7 +103,8 @@ class CallManager(
    private val discoveredCalleePeers = mutableSetOf<HexKey>()

    companion object {
-        const val CALL_TIMEOUT_MS = 60_000L // 60 seconds ringing timeout
+        const val CALL_TIMEOUT_MS = 60_000L // 60 seconds ringing timeout (callee side)
+        const val PEER_INVITE_TIMEOUT_MS = 30_000L // 30 seconds per-peer invite timeout (caller side)
        const val ENDED_DISPLAY_MS = 2_000L // show "call ended" briefly before resetting
        const val MAX_EVENT_AGE_SECONDS = 20L // discard signaling events older than this
        const val MAX_PROCESSED_EVENT_IDS = 2_000 // cap dedup set to prevent unbounded growth
@@ -140,7 +149,8 @@ class CallManager(
        callType: CallType,
    ) = stateMutex.withLock {
        _state.value = CallState.Offering(callId, calleePubKeys, callType)
-        startTimeout(callId)
+        cancelAllPeerTimeouts()
+        calleePubKeys.forEach { schedulePeerTimeout(it, callId) }
    }

    /**
@@ -201,10 +211,11 @@ class CallManager(
        val result = factory.createCallOffer(sdpOffer, calleePubKey, callId, callType, signer)
        stateMutex.withLock {
            _state.value = CallState.Offering(callId, setOf(calleePubKey), callType)
-            startTimeout(callId)
+            cancelAllPeerTimeouts()
+            schedulePeerTimeout(calleePubKey, callId)
        }
        publishEvent(result.wrap)
-        Log.d("CallManager") { "initiateCall: offer published, timeout started" }
+        Log.d("CallManager") { "initiateCall: offer published, per-peer timeout started" }
    }

    // ---- Incoming call handling ----
@@ -348,7 +359,9 @@ class CallManager(
                        current.callType,
                        pendingPeerPubKeys = pending,
                    )
-                cancelTimeout()
+                // The answered peer no longer needs its invite timer. Peers
+                // still in `pending` keep theirs (scheduled in beginOffering).
+                cancelPeerTimeout(answeringPeer)
                Log.d("CallManager") { "onCallAnswered: Offering -> Connecting, forwarding answer to CallController" }
                onAnswerReceived?.invoke(event)
            }
@@ -362,6 +375,7 @@ class CallManager(
                                peerPubKeys = current.peerPubKeys + answeringPeer,
                                pendingPeerPubKeys = current.pendingPeerPubKeys - answeringPeer,
                            )
+                        cancelPeerTimeout(answeringPeer)
                    }

                    answeringPeer !in current.peerPubKeys -> {
@@ -398,6 +412,7 @@ class CallManager(
                                peerPubKeys = current.peerPubKeys + answeringPeer,
                                pendingPeerPubKeys = current.pendingPeerPubKeys - answeringPeer,
                            )
+                        cancelPeerTimeout(answeringPeer)
                    }

                    answeringPeer !in current.peerPubKeys -> {
@@ -433,6 +448,7 @@ class CallManager(
        when (current) {
            is CallState.Offering -> {
                if (callId != current.callId) return
+                cancelPeerTimeout(rejectingPeer)
                val remaining = current.peerPubKeys - rejectingPeer
                if (remaining.isEmpty()) {
                    transitionToEnded(current.callId, current.peerPubKeys, EndReason.PEER_REJECTED)
@@ -444,6 +460,7 @@ class CallManager(

            is CallState.Connecting -> {
                if (callId != current.callId) return
+                cancelPeerTimeout(rejectingPeer)
                _state.value =
                    current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer)
                onPeerLeft?.invoke(rejectingPeer)
@@ -451,6 +468,7 @@ class CallManager(

            is CallState.Connected -> {
                if (callId != current.callId) return
+                cancelPeerTimeout(rejectingPeer)
                _state.value =
                    current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer)
                onPeerLeft?.invoke(rejectingPeer)
@@ -567,6 +585,10 @@ class CallManager(
            }
        }

+        // Start the per-peer invite timer. If the invitee does not answer
+        // within PEER_INVITE_TIMEOUT_MS, they are dropped from the call.
+        schedulePeerTimeout(peerPubKey, callId)
+
        val allMembers = existingMembers + peerPubKey + signer.pubKey
        val result = factory.createCallOffer(sdpOffer, peerPubKey, allMembers, callId, callType, signer)
        publishEvent(result.wrap)
@@ -616,6 +638,7 @@ class CallManager(
        when (current) {
            is CallState.Connected -> {
                if (callId != current.callId) return
+                cancelPeerTimeout(leavingPeer)
                val connectedRemaining = current.peerPubKeys - leavingPeer
                val pendingRemaining = current.pendingPeerPubKeys - leavingPeer
                if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) {
@@ -634,6 +657,7 @@ class CallManager(

            is CallState.Connecting -> {
                if (callId != current.callId) return
+                cancelPeerTimeout(leavingPeer)
                val connectedRemaining = current.peerPubKeys - leavingPeer
                val pendingRemaining = current.pendingPeerPubKeys - leavingPeer
                if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) {
@@ -652,6 +676,7 @@ class CallManager(

            is CallState.Offering -> {
                if (callId != current.callId) return
+                cancelPeerTimeout(leavingPeer)
                val remaining = current.peerPubKeys - leavingPeer
                if (remaining.isEmpty()) {
                    transitionToEnded(callId, current.peerPubKeys, EndReason.PEER_HANGUP)
@@ -763,6 +788,7 @@ class CallManager(
    fun reset() {
        _state.value = CallState.Idle
        cancelTimeout()
+        cancelAllPeerTimeouts()
        resetJob?.cancel()
        resetJob = null
        processedEventIds.clear()
@@ -780,6 +806,7 @@ class CallManager(
        discoveredCalleePeers.clear()
        _state.value = CallState.Ended(callId, peerPubKeys, reason)
        cancelTimeout()
+        cancelAllPeerTimeouts()
        resetJob?.cancel()
        resetJob =
            scope.launch {
@@ -791,6 +818,118 @@ class CallManager(
            }
    }

+    // ---- Per-peer invite timeout ----
+
+    /**
+     * Starts a 30-second timer for [peerPubKey]. If the peer has not answered
+     * by the time it fires, [handlePeerTimeout] drops them from the current
+     * group call and publishes a CallHangup to them so their device stops
+     * ringing.
+     *
+     * Safe to call multiple times for the same peer — any previous timer is
+     * cancelled first.
+     */
+    private fun schedulePeerTimeout(
+        peerPubKey: HexKey,
+        callId: String,
+    ) {
+        perPeerTimeoutJobs.remove(peerPubKey)?.cancel()
+        perPeerTimeoutJobs[peerPubKey] =
+            scope.launch {
+                delay(PEER_INVITE_TIMEOUT_MS)
+                handlePeerTimeout(peerPubKey, callId)
+            }
+    }
+
+    /** Cancels the per-peer timer for [peerPubKey], if any. */
+    private fun cancelPeerTimeout(peerPubKey: HexKey) {
+        perPeerTimeoutJobs.remove(peerPubKey)?.cancel()
+    }
+
+    /** Cancels every per-peer timer. Called on terminal state transitions. */
+    private fun cancelAllPeerTimeouts() {
+        perPeerTimeoutJobs.values.forEach { it.cancel() }
+        perPeerTimeoutJobs.clear()
+    }
+
+    /**
+     * Handles a per-peer timeout firing. Drops the peer from the current
+     * group call state and publishes a CallHangup to them.
+     *
+     * - In [CallState.Offering] the peer is removed from `peerPubKeys`; if no
+     *   peers remain, the whole call ends with [EndReason.TIMEOUT].
+     * - In [CallState.Connecting] the peer is removed from `pendingPeerPubKeys`;
+     *   if that leaves nobody connected AND no more pending, the call ends
+     *   with [EndReason.TIMEOUT]. Otherwise the call continues with the
+     *   already-connected peers.
+     * - In [CallState.Connected] the peer is removed from `pendingPeerPubKeys`;
+     *   at least one other peer is connected by definition, so the call
+     *   always continues.
+     *
+     * Fires [onPeerLeft] so the CallController disposes the per-peer
+     * PeerConnection (and any pending ICE buffers) for the dropped peer.
+     */
+    private suspend fun handlePeerTimeout(
+        peerPubKey: HexKey,
+        callId: String,
+    ) {
+        var shouldPublishHangup = false
+        stateMutex.withLock {
+            perPeerTimeoutJobs.remove(peerPubKey)
+            when (val current = _state.value) {
+                is CallState.Offering -> {
+                    if (callId != current.callId) return@withLock
+                    if (peerPubKey !in current.peerPubKeys) return@withLock
+                    Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Offering" }
+                    shouldPublishHangup = true
+                    val remaining = current.peerPubKeys - peerPubKey
+                    if (remaining.isEmpty()) {
+                        transitionToEnded(current.callId, current.peerPubKeys, EndReason.TIMEOUT)
+                    } else {
+                        _state.value = current.copy(peerPubKeys = remaining)
+                        onPeerLeft?.invoke(peerPubKey)
+                    }
+                }
+
+                is CallState.Connecting -> {
+                    if (callId != current.callId) return@withLock
+                    if (peerPubKey !in current.pendingPeerPubKeys) return@withLock
+                    Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Connecting" }
+                    shouldPublishHangup = true
+                    val newPending = current.pendingPeerPubKeys - peerPubKey
+                    if (current.peerPubKeys.isEmpty() && newPending.isEmpty()) {
+                        transitionToEnded(
+                            current.callId,
+                            current.peerPubKeys + current.pendingPeerPubKeys,
+                            EndReason.TIMEOUT,
+                        )
+                    } else {
+                        _state.value = current.copy(pendingPeerPubKeys = newPending)
+                        onPeerLeft?.invoke(peerPubKey)
+                    }
+                }
+
+                is CallState.Connected -> {
+                    if (callId != current.callId) return@withLock
+                    if (peerPubKey !in current.pendingPeerPubKeys) return@withLock
+                    Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Connected" }
+                    shouldPublishHangup = true
+                    _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - peerPubKey)
+                    onPeerLeft?.invoke(peerPubKey)
+                }
+
+                else -> {
+                    return@withLock
+                }
+            }
+        }
+
+        if (shouldPublishHangup) {
+            val result = factory.createHangup(peerPubKey, callId, signer = signer)
+            publishEvent(result.wrap)
+        }
+    }
+
    private fun startTimeout(callId: String) {
        cancelTimeout()
        timeoutJob =
@@ -34,6 +34,7 @@ import com.vitorpamplona.quartz.nipACWebRtcCalls.tags.CallType
 import com.vitorpamplona.quartz.utils.TimeUtils
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.advanceTimeBy
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
@@ -1291,6 +1292,185 @@ class CallManagerTest {
            assertEquals(carol, bobForwardedAnswer, "Bob must forward Carol's answer to his CallController")
        }

+    // ========================================================================
+    // Per-peer 30-second invite timeout
+    // ========================================================================
+
+    /**
+     * P2P call: Alice calls Bob, Bob never answers. After 30 s Alice's
+     * per-peer timer fires and the call ends with [EndReason.TIMEOUT]. The
+     * timeout hangup is also published so Bob's device stops ringing.
+     */
+    @Test
+    fun perPeerTimeoutEndsP2PCallWhenBobNeverAnswers() =
+        runTest {
+            val (manager, published) = createManager(localPubKey = alice, followedKeys = setOf(bob))
+
+            manager.initiateCall(bob, CallType.VOICE, callId, sdpOffer)
+            assertIs<CallState.Offering>(manager.state.value)
+            published.clear()
+
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS + 100)
+
+            val ended = manager.state.value
+            assertIs<CallState.Ended>(ended)
+            assertEquals(EndReason.TIMEOUT, ended.reason)
+            assertEquals(
+                1,
+                published.size,
+                "Timeout should publish exactly one hangup to the unresponsive peer",
+            )
+        }
+
+    /**
+     * Group call: Alice offers to Bob + Carol. Bob answers quickly, Carol
+     * does not. After 30 s Alice's timer for Carol fires and Carol is
+     * removed from pending. The call continues with Bob. A hangup is
+     * published to Carol but the Bob leg is untouched.
+     */
+    @Test
+    fun perPeerTimeoutDropsSlowCalleeFromGroupCall() =
+        runTest {
+            val (manager, published) = createManager(localPubKey = alice, followedKeys = setOf(bob, carol))
+
+            manager.beginOffering(callId, setOf(bob, carol), CallType.VOICE)
+            assertIs<CallState.Offering>(manager.state.value)
+
+            // Bob answers quickly (well before the 30 s timer).
+            manager.onSignalingEvent(makeGroupAnswer(from = bob, members = setOf(alice, bob, carol)))
+            val afterBob = manager.state.value
+            assertIs<CallState.Connecting>(afterBob)
+            assertTrue(bob in afterBob.peerPubKeys)
+            assertTrue(carol in afterBob.pendingPeerPubKeys)
+            published.clear()
+
+            // Advance past the 30 s per-peer timeout.
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS + 100)
+
+            // Carol dropped; call continues with Bob.
+            val state = manager.state.value
+            assertIs<CallState.Connecting>(state)
+            assertTrue(carol !in state.pendingPeerPubKeys, "Carol should be dropped from pending")
+            assertTrue(bob in state.peerPubKeys, "Bob leg must be untouched")
+
+            assertEquals(
+                1,
+                published.size,
+                "Timeout must publish exactly one hangup (addressed to Carol)",
+            )
+        }
+
+    /**
+     * Bob answering inside the 30 s window cancels his per-peer timer.
+     * Advancing 60 s afterwards must not fire any phantom timeout.
+     */
+    @Test
+    fun perPeerTimeoutIsCancelledOnAnswer() =
+        runTest {
+            val (manager, published) = createManager(localPubKey = alice, followedKeys = setOf(bob))
+
+            manager.initiateCall(bob, CallType.VOICE, callId, sdpOffer)
+            assertIs<CallState.Offering>(manager.state.value)
+
+            // Bob answers before the timeout fires.
+            manager.onSignalingEvent(makeAnswer(from = bob, to = alice))
+            assertIs<CallState.Connecting>(manager.state.value)
+            manager.onPeerConnected()
+            assertIs<CallState.Connected>(manager.state.value)
+            published.clear()
+
+            // Advance way past the 30 s timeout — no timeout must fire.
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS * 2)
+
+            assertIs<CallState.Connected>(manager.state.value)
+            assertTrue(published.isEmpty(), "No timeout hangup must be published after a successful answer")
+        }
+
+    /**
+     * Mid-call invite: Alice is Connected with Bob, then invites Carol.
+     * Carol never answers. After 30 s Carol is dropped from
+     * [CallState.Connected.pendingPeerPubKeys] and the call continues with
+     * Bob. A hangup is published to Carol.
+     */
+    @Test
+    fun perPeerTimeoutDropsMidCallInviteeWhenNoAnswer() =
+        runTest {
+            val (manager, published) = createManager(localPubKey = alice, followedKeys = setOf(bob))
+
+            // Alice ↔ Bob Connected.
+            manager.initiateCall(bob, CallType.VOICE, callId, sdpOffer)
+            manager.onSignalingEvent(makeAnswer(from = bob, to = alice))
+            manager.onPeerConnected()
+            assertIs<CallState.Connected>(manager.state.value)
+
+            // Alice invites Carol.
+            manager.invitePeer(carol, "invite-sdp")
+            val afterInvite = manager.state.value
+            assertIs<CallState.Connected>(afterInvite)
+            assertTrue(carol in afterInvite.pendingPeerPubKeys)
+            published.clear()
+
+            // Carol never answers — advance past the 30 s invite timeout.
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS + 100)
+
+            val state = manager.state.value
+            assertIs<CallState.Connected>(state)
+            assertTrue(carol !in state.pendingPeerPubKeys, "Carol should be dropped from pending")
+            assertTrue(bob in state.peerPubKeys, "Bob leg must be untouched")
+
+            assertEquals(
+                1,
+                published.size,
+                "Invite timeout must publish exactly one hangup addressed to Carol",
+            )
+        }
+
+    /**
+     * Bob rejecting a P2P offer cancels his per-peer timer so advancing time
+     * past 30 s afterwards does not publish a duplicate hangup on top of the
+     * Ended transition.
+     */
+    @Test
+    fun perPeerTimeoutIsCancelledOnReject() =
+        runTest {
+            val (manager, published) = createManager(localPubKey = alice, followedKeys = setOf(bob))
+
+            manager.initiateCall(bob, CallType.VOICE, callId, sdpOffer)
+            manager.onSignalingEvent(makeReject(from = bob, to = alice))
+            val ended = manager.state.value
+            assertIs<CallState.Ended>(ended)
+            assertEquals(EndReason.PEER_REJECTED, ended.reason)
+
+            val publishedAfterReject = published.size
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS * 2)
+            assertEquals(
+                publishedAfterReject,
+                published.size,
+                "No timeout hangup should be published after the peer rejected",
+            )
+        }
+
+    /**
+     * When a group offer is made, each callee gets its own timer. If NEITHER
+     * answers, both timers fire in turn and the whole call ends with
+     * [EndReason.TIMEOUT] (the second timeout has no peers left, so it tips
+     * the state over the edge).
+     */
+    @Test
+    fun perPeerTimeoutEndsCallWhenAllCalleesIgnore() =
+        runTest {
+            val (manager, _) = createManager(localPubKey = alice, followedKeys = setOf(bob, carol))
+
+            manager.beginOffering(callId, setOf(bob, carol), CallType.VOICE)
+            assertIs<CallState.Offering>(manager.state.value)
+
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS + 100)
+
+            val ended = manager.state.value
+            assertIs<CallState.Ended>(ended)
+            assertEquals(EndReason.TIMEOUT, ended.reason)
+        }
+
    @Test
    fun interfaceFullP2PCallFlowWithRealSigners() =
        runTest {