diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt index 44bbe8d96..2f958704e 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt @@ -96,6 +96,18 @@ class CallManager( * whole call ends. */ private val perPeerTimeoutJobs = mutableMapOf() + /** 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() + /** 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) } diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt index 0c487cc0d..f9da7138a 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt @@ -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(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(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(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(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(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(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(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(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(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(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(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(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