From f670724cd807363af8a9cdb57e1bbc6b0289f166 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 15 Apr 2026 14:36:02 +0000 Subject: [PATCH] fix: drop unresponsive group-call peers on callee side after 30 s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a caller invites a group and one of the members never answers, the caller already drops the unresponsive peer after 30 s (via the per-peer watchdog in CallManager). The hangup the caller publishes is addressed only to the unresponsive peer, so the OTHER participants never learn that the peer is out and keep waiting for them forever. Mirror the caller's 30 s budget on the callee side. In acceptCall, split the group members into: - peerPubKeys: the caller (they sent us the offer) plus any peers whose answer we observed while still ringing. These are confirmed to be in the call. - pendingPeerPubKeys: everyone else we haven't heard from yet. Each gets a local watchdog timer; if we never observe them (via a mesh answer or a mid-call offer) within 30 s they are silently dropped from our local state. Because the callee is not the peer's inviter, handlePeerTimeout must NOT publish a CallHangup in this path — terminating the invitee's ringing is the caller's responsibility. Track "who we invited" in a peersInvitedByUs set so the caller-side path (beginOffering, initiateCall, invitePeer) still publishes the hangup, while the new callee-side watchdog path drops the peer silently. Also move a pending peer into peerPubKeys and cancel their watchdog when they send us a mid-call mesh offer — the offer is proof they're in the call, so we should not wait further. Adds three new tests (callee watchdog drops pending peer, mid-call offer transitions pending peer, watchdog is cancelled on answer from pending peer) and updates two existing tests for the new Connecting state shape when a group call is accepted. --- .../amethyst/commons/call/CallManager.kt | 96 +++++++++++- .../amethyst/commons/call/CallManagerTest.kt | 139 +++++++++++++++++- 2 files changed, 220 insertions(+), 15 deletions(-) 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