Merge branch 'main' of https://github.com/vitorpamplona/amethyst

2026-04-15 10:51:18 -04:00
parent 5f4149d3dc 31ad5839d7
commit 99c5f08eb3
2 changed files with 220 additions and 15 deletions
@@ -96,6 +96,18 @@ class CallManager(
     *  whole call ends. */
    private val perPeerTimeoutJobs = mutableMapOf<HexKey, Job>()

+    /** Peers we have published a CallOffer to (either as the call initiator
+     *  or via a mid-call invitePeer). When their per-peer timer fires we
+     *  publish a CallHangup so their device stops ringing.
+     *
+     *  Peers we are merely *waiting* on (i.e. members of a group call we
+     *  accepted but who haven't yet joined via mesh) are tracked only with
+     *  their timer job — NOT in this set. When their timer fires we drop
+     *  them silently from local state without sending any further
+     *  signaling, because their ringing is the caller's responsibility to
+     *  terminate. */
+    private val peersInvitedByUs = mutableSetOf<HexKey>()
+
    /** 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
@@ -159,6 +171,7 @@ class CallManager(
    ) = stateMutex.withLock {
        _state.value = CallState.Offering(callId, calleePubKeys, callType)
        cancelAllPeerTimeouts()
+        peersInvitedByUs.addAll(calleePubKeys)
        calleePubKeys.forEach { schedulePeerTimeout(it, callId) }
    }

@@ -221,6 +234,7 @@ class CallManager(
        stateMutex.withLock {
            _state.value = CallState.Offering(callId, setOf(calleePubKey), callType)
            cancelAllPeerTimeouts()
+            peersInvitedByUs.add(calleePubKey)
            schedulePeerTimeout(calleePubKey, callId)
        }
        publishEvent(result.wrap)
@@ -266,6 +280,35 @@ class CallManager(
                }
            if (callId == currentCallId) {
                Log.d("CallManager") { "Mid-call offer from ${callerPubKey.take(8)} for current call — callee-to-callee" }
+                // If this peer was in our pending set (we were waiting for
+                // them to join), move them into the connected set and
+                // cancel their watchdog timer — the offer is proof they're
+                // in the call.
+                when (currentState) {
+                    is CallState.Connecting -> {
+                        if (callerPubKey in currentState.pendingPeerPubKeys) {
+                            _state.value =
+                                currentState.copy(
+                                    peerPubKeys = currentState.peerPubKeys + callerPubKey,
+                                    pendingPeerPubKeys = currentState.pendingPeerPubKeys - callerPubKey,
+                                )
+                            cancelPeerTimeout(callerPubKey)
+                        }
+                    }
+
+                    is CallState.Connected -> {
+                        if (callerPubKey in currentState.pendingPeerPubKeys) {
+                            _state.value =
+                                currentState.copy(
+                                    peerPubKeys = currentState.peerPubKeys + callerPubKey,
+                                    pendingPeerPubKeys = currentState.pendingPeerPubKeys - callerPubKey,
+                                )
+                            cancelPeerTimeout(callerPubKey)
+                        }
+                    }
+
+                    else -> {}
+                }
                onMidCallOfferReceived?.invoke(callerPubKey, event.sdpOffer())
                return
            }
@@ -303,11 +346,39 @@ class CallManager(
            }
            current = s

-            Log.d("CallManager") { "acceptCall: callId=${current.callId}, transitioning to Connecting, sdpAnswerLength=${sdpAnswer.length}" }
-            _state.value = CallState.Connecting(current.callId, current.peerPubKeys() - signer.pubKey, current.callType)
-            cancelTimeout()
            discovered = discoveredCalleePeers.toSet()
            discoveredCalleePeers.clear()
+
+            // Split group members into "known to be in the call" vs "still
+            // waiting on". The caller is definitely in the call (they sent
+            // us the offer). Peers whose answer we observed while ringing
+            // are also confirmed. Everyone else is placed in
+            // pendingPeerPubKeys with a per-peer watchdog timer so that
+            // unresponsive group members are dropped from our local UI
+            // after the same 30-second budget the caller uses — otherwise
+            // we would wait for them forever (the caller's timeout hangup
+            // is addressed only to the unresponsive peer and never reaches
+            // the rest of us).
+            val knownPeers = (discovered + current.callerPubKey) - signer.pubKey
+            val pendingOnJoin = (current.peerPubKeys() - signer.pubKey) - knownPeers
+
+            Log.d("CallManager") {
+                "acceptCall: callId=${current.callId}, transitioning to Connecting, " +
+                    "known=${knownPeers.size}, pending=${pendingOnJoin.size}, " +
+                    "sdpAnswerLength=${sdpAnswer.length}"
+            }
+            _state.value =
+                CallState.Connecting(
+                    callId = current.callId,
+                    peerPubKeys = knownPeers,
+                    callType = current.callType,
+                    pendingPeerPubKeys = pendingOnJoin,
+                )
+            cancelTimeout()
+            // Local watchdog timers only — we are not the inviter for these
+            // peers, so [handlePeerTimeout] will silently drop them without
+            // publishing any hangup (see `peersInvitedByUs`).
+            pendingOnJoin.forEach { schedulePeerTimeout(it, current.callId) }
        }

        val allMembers = current.groupMembers + signer.pubKey
@@ -605,6 +676,8 @@ 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.
+        // We sent the offer, so a timeout must also publish a hangup.
+        peersInvitedByUs.add(peerPubKey)
        schedulePeerTimeout(peerPubKey, callId)

        val allMembers = existingMembers + peerPubKey + signer.pubKey
@@ -862,12 +935,14 @@ class CallManager(
    /** Cancels the per-peer timer for [peerPubKey], if any. */
    private fun cancelPeerTimeout(peerPubKey: HexKey) {
        perPeerTimeoutJobs.remove(peerPubKey)?.cancel()
+        peersInvitedByUs.remove(peerPubKey)
    }

    /** Cancels every per-peer timer. Called on terminal state transitions. */
    private fun cancelAllPeerTimeouts() {
        perPeerTimeoutJobs.values.forEach { it.cancel() }
        perPeerTimeoutJobs.clear()
+        peersInvitedByUs.clear()
    }

    /**
@@ -894,12 +969,17 @@ class CallManager(
        var shouldPublishHangup = false
        stateMutex.withLock {
            perPeerTimeoutJobs.remove(peerPubKey)
+            // Only publish a hangup if we were the peer's inviter. Pure
+            // "watchdog" timers started by a callee for group members who
+            // never joined must NOT publish anything — terminating the
+            // invitee's ringing is the caller's responsibility.
+            val wasInvitedByUs = peersInvitedByUs.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
+                    shouldPublishHangup = wasInvitedByUs
                    val remaining = current.peerPubKeys - peerPubKey
                    if (remaining.isEmpty()) {
                        transitionToEnded(current.callId, current.peerPubKeys, EndReason.TIMEOUT)
@@ -912,8 +992,8 @@ class CallManager(
                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
+                    Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Connecting (invitedByUs=$wasInvitedByUs)" }
+                    shouldPublishHangup = wasInvitedByUs
                    val newPending = current.pendingPeerPubKeys - peerPubKey
                    if (current.peerPubKeys.isEmpty() && newPending.isEmpty()) {
                        transitionToEnded(
@@ -930,8 +1010,8 @@ class CallManager(
                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
+                    Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Connected (invitedByUs=$wasInvitedByUs)" }
+                    shouldPublishHangup = wasInvitedByUs
                    _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - peerPubKey)
                    onPeerLeft?.invoke(peerPubKey)
                }
@@ -1162,8 +1162,9 @@ class CallManagerTest {
     * Regression: in an initial group call, the callees observing each other's
     * answers MUST NOT trip the mid-call expansion branch. The answering peer
     * was already part of the group membership set by [acceptCall] (via the
-     * IncomingCall.groupMembers → Connecting.peerPubKeys transition), so no
-     * additional insertion should occur.
+     * IncomingCall.groupMembers → Connecting.peerPubKeys/pendingPeerPubKeys
+     * transition), so the combined size must stay the same — the peer only
+     * moves out of pending into connected.
     */
    @Test
    fun initialCallAnswerFromKnownPeerDoesNotExpandMembership() =
@@ -1174,14 +1175,18 @@ class CallManagerTest {
            manager.onSignalingEvent(makeGroupOffer(from = alice, members = setOf(bob, carol)))
            assertIs<CallState.IncomingCall>(manager.state.value)

-            // Bob accepts. State becomes Connecting(peerPubKeys={alice, carol}).
+            // Bob accepts. State becomes
+            // Connecting(peerPubKeys={alice}, pendingPeerPubKeys={carol}).
            manager.acceptCall(sdpAnswer)
            val connecting = manager.state.value
            assertIs<CallState.Connecting>(connecting)
            assertTrue(alice in connecting.peerPubKeys)
-            assertTrue(carol in connecting.peerPubKeys)
+            assertTrue(
+                carol in connecting.pendingPeerPubKeys,
+                "Carol should start in pending until Bob observes her joining",
+            )

-            val sizeBefore = connecting.peerPubKeys.size
+            val totalBefore = connecting.peerPubKeys.size + connecting.pendingPeerPubKeys.size

            // Carol — who is already in Bob's tracked membership — answers.
            // This is the normal initial-call mesh observation path.
@@ -1189,7 +1194,10 @@ class CallManagerTest {

            val after = manager.state.value
            assertIs<CallState.Connecting>(after)
-            assertEquals(sizeBefore, after.peerPubKeys.size, "Known peer's answer must not grow peerPubKeys")
+            val totalAfter = after.peerPubKeys.size + after.pendingPeerPubKeys.size
+            assertEquals(totalBefore, totalAfter, "Known peer's answer must not grow the tracked membership")
+            assertTrue(carol in after.peerPubKeys, "Carol should move into peerPubKeys on answer")
+            assertTrue(carol !in after.pendingPeerPubKeys, "Carol should leave pendingPeerPubKeys on answer")
        }

    /**
@@ -1261,11 +1269,17 @@ class CallManagerTest {

            // Step 4: Carol accepts. Her Connecting state must include Bob
            // (so later mid-call offers from Bob are handled correctly).
+            // Alice (the caller) is placed directly into peerPubKeys; Bob
+            // is placed into pendingPeerPubKeys with a local watchdog
+            // timer until we observe him in the call.
            carolManager.acceptCall("carol-answer-sdp")
            val carolConnecting = carolManager.state.value
            assertIs<CallState.Connecting>(carolConnecting)
-            assertTrue(bob in carolConnecting.peerPubKeys, "Carol's Connecting state must include Bob as a peer")
            assertTrue(alice in carolConnecting.peerPubKeys, "Carol's Connecting state must include Alice as a peer")
+            assertTrue(
+                bob in carolConnecting.pendingPeerPubKeys,
+                "Carol's Connecting state must track Bob as a pending peer until he joins",
+            )

            // Step 5: Alice receives Carol's answer broadcast. Carol moves
            // out of pending into peerPubKeys.
@@ -1452,6 +1466,117 @@ class CallManagerTest {
            )
        }

+    /**
+     * Callee-side group call: Bob accepts a group offer from Alice that
+     * also included Carol. Carol never joins. After 30 s Bob's local
+     * watchdog timer drops Carol from his Connecting state WITHOUT
+     * publishing any signaling (the caller is responsible for terminating
+     * Carol's ringing — not Bob).
+     */
+    @Test
+    fun perPeerTimeoutDropsUnresponsiveGroupMemberOnCalleeSide() =
+        runTest {
+            val (manager, published) = createManager(localPubKey = bob, followedKeys = setOf(alice, carol))
+
+            // Alice calls {bob, carol} as a group. Bob receives and accepts.
+            manager.onSignalingEvent(makeGroupOffer(from = alice, members = setOf(bob, carol)))
+            assertIs<CallState.IncomingCall>(manager.state.value)
+            manager.acceptCall(sdpAnswer)
+
+            // Bob's Connecting state: Alice (caller) is connected, Carol
+            // is pending with a watchdog timer.
+            val connecting = manager.state.value
+            assertIs<CallState.Connecting>(connecting)
+            assertTrue(alice in connecting.peerPubKeys, "Caller must be in peerPubKeys")
+            assertTrue(carol in connecting.pendingPeerPubKeys, "Unseen peer must be pending")
+            published.clear()
+
+            // Advance past the 30 s per-peer timeout. Carol never sent an
+            // answer or a mesh offer.
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS + 100)
+
+            val state = manager.state.value
+            assertIs<CallState.Connecting>(state)
+            assertTrue(carol !in state.pendingPeerPubKeys, "Carol must be dropped from pending")
+            assertTrue(carol !in state.peerPubKeys, "Carol must not be silently promoted")
+            assertTrue(alice in state.peerPubKeys, "Alice leg must be untouched")
+
+            // The callee MUST NOT publish a hangup — Carol is still
+            // connected to Alice from the caller's perspective, and
+            // terminating her ringing is Alice's responsibility.
+            assertTrue(
+                published.isEmpty(),
+                "Callee watchdog timeout must not publish any signaling",
+            )
+        }
+
+    /**
+     * Callee-side group call: Bob accepts a group offer that included
+     * Carol. Carol publishes a mesh offer to Bob mid-call. Bob must move
+     * Carol out of pending into peerPubKeys and cancel her watchdog timer
+     * — the offer is proof she's in the call.
+     */
+    @Test
+    fun midCallOfferFromPendingPeerMovesThemIntoConnected() =
+        runTest {
+            val (manager, _) = createManager(localPubKey = bob, followedKeys = setOf(alice, carol))
+
+            manager.onSignalingEvent(makeGroupOffer(from = alice, members = setOf(bob, carol)))
+            manager.acceptCall(sdpAnswer)
+            val connecting = manager.state.value
+            assertIs<CallState.Connecting>(connecting)
+            assertTrue(carol in connecting.pendingPeerPubKeys)
+
+            // Carol publishes a mesh offer to Bob (same call-id).
+            var midCallOfferPeer: HexKey? = null
+            manager.onMidCallOfferReceived = { peer, _ -> midCallOfferPeer = peer }
+            manager.onSignalingEvent(makeGroupOffer(from = carol, members = setOf(alice, bob, carol)))
+
+            val after = manager.state.value
+            assertIs<CallState.Connecting>(after)
+            assertTrue(carol in after.peerPubKeys, "Carol should move into peerPubKeys")
+            assertTrue(carol !in after.pendingPeerPubKeys, "Carol should leave pending")
+            assertEquals(carol, midCallOfferPeer, "CallController must still receive the mid-call offer")
+
+            // And her watchdog should be cancelled — advancing past the
+            // 30 s budget must not further alter state.
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS + 100)
+            val later = manager.state.value
+            assertIs<CallState.Connecting>(later)
+            assertTrue(carol in later.peerPubKeys, "Carol must remain connected after timeout budget")
+        }
+
+    /**
+     * Regression: the callee-side watchdog must be cancelled when a
+     * pending peer's answer arrives, so advancing past 30 s afterwards
+     * does not cause a phantom state change or publish.
+     */
+    @Test
+    fun calleeWatchdogIsCancelledOnAnswerFromPendingPeer() =
+        runTest {
+            val (manager, published) = createManager(localPubKey = bob, followedKeys = setOf(alice, carol))
+
+            manager.onSignalingEvent(makeGroupOffer(from = alice, members = setOf(bob, carol)))
+            manager.acceptCall(sdpAnswer)
+            assertTrue(carol in (manager.state.value as CallState.Connecting).pendingPeerPubKeys)
+
+            // Carol's answer reaches Bob well before her watchdog fires.
+            manager.onSignalingEvent(makeGroupAnswer(from = carol, members = setOf(alice, bob, carol)))
+            val afterAnswer = manager.state.value
+            assertIs<CallState.Connecting>(afterAnswer)
+            assertTrue(carol in afterAnswer.peerPubKeys)
+            assertTrue(carol !in afterAnswer.pendingPeerPubKeys)
+            published.clear()
+
+            // Advance past the 30 s watchdog — nothing should fire.
+            advanceTimeBy(CallManager.PEER_INVITE_TIMEOUT_MS * 2)
+            assertIs<CallState.Connecting>(manager.state.value)
+            assertTrue(
+                published.isEmpty(),
+                "Cancelled watchdog must not publish a hangup after the peer answered",
+            )
+        }
+
    /**
     * 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