nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: restore onDestroy hangup, prevent event drops, fix deadlock risk

Browse Source 
Three fixes from audit:

1. PiP swipe-away regression (Bug 5): Restored hangup signaling in
   CallActivity.onDestroy via detached CoroutineScope. This is the
   PRIMARY signaling path for PiP dismiss, back press, and finish().
   CallForegroundService.onTaskRemoved is the BACKUP for task swipe
   from Recents. Both paths are idempotent — double-publishing is safe.

2. SharedFlow event drops (Bug 1): Increased sessionEvents buffer from
   16 to 256 to handle worst-case ICE candidate bursts. Added
   emitSessionEvent() helper that logs drops as errors. Increased
   renegotiationEvents buffer from 8 to 32. tryEmit is kept (not
   emit) to avoid suspending while holding stateMutex, which would
   block all signaling.

3. runBlocking deadlock risk (Bug 2): CallSessionBridge.clear() now
   uses CallManager.reset() (non-blocking, no mutex) instead of
   runBlocking { hangup() }. The bridge teardown is for local state
   cleanup during account switch — hangup signaling is the Activity's
   and Service's responsibility.

https://claude.ai/code/session_019yNnDjGKmJb19gadmojq54
This commit is contained in:
Claude
2026-04-16 15:03:50 +00:00
parent 89cf5e14d4
commit 29c5c26af7
3 changed files with 78 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallSessionBridge.kt
+10 -25
View File
@@ -21,11 +21,7 @@
package com.vitorpamplona.amethyst.service.call
import com.vitorpamplona.amethyst.commons.call.CallManager
import com.vitorpamplona.amethyst.commons.call.CallState
import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
/**
* Process-level singleton that bridges the active [CallManager] and
@@ -52,29 +48,18 @@ object CallSessionBridge {
}
/**
* Terminates any active call and clears all references. Called from
* [AccountViewModel.onCleared] during logout or account switch so
* that a stale CallSession cannot invoke signing/publishing lambdas
* on a disposed Account.
* Resets call state and clears all references. Called from
* [AccountViewModel.onCleared] during logout or account switch.
*
* Uses [CallManager.reset] (non-blocking, no mutex) instead of
* [CallManager.hangup] to avoid deadlocking on `stateMutex` if
* a cancelled coroutine on the dying `viewModelScope` still holds
* it. Hangup signaling to the remote peer is the responsibility
* of [CallActivity.onDestroy] and [CallForegroundService], not
* the bridge teardown.
*/
fun clear() {
val mgr = callManager
if (mgr != null) {
val state = mgr.state.value
if (state is CallState.IncomingCall ||
state is CallState.Offering ||
state is CallState.Connecting ||
state is CallState.Connected
) {
try {
runBlocking {
withTimeoutOrNull(3_000L) { mgr.hangup() }
}
} catch (e: Exception) {
Log.e("CallSessionBridge", "clear: hangup failed", e)
}
}
}
callManager?.reset()
callManager = null
accountViewModel = null
}
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/CallActivity.kt
+36 -5
View File
@@ -231,11 +231,42 @@ class CallActivity : AppCompatActivity() {
session?.close()
session = null
// Hangup signaling is NOT published here. It is handled by
// CallForegroundService.onTaskRemoved() / onDestroy() which
// runs synchronously via runBlocking with a 3-second timeout.
// This is more reliable than the previous fire-and-forget
// CoroutineScope that could be killed before completing.
// Publish hangup/reject so the remote side stops ringing.
// This is the PRIMARY signaling path (covers PiP dismiss,
// back press, finish()). CallForegroundService.onTaskRemoved
// is the BACKUP for task-swipe-from-Recents where Activity
// onDestroy may not complete in time.
//
// hangup()/rejectCall() are idempotent — if the state is
// already Ended/Idle from a prior transition, they return
// immediately. So double-publishing from both Activity and
// Service is safe.
val manager = CallSessionBridge.callManager
if (manager != null) {
when (manager.state.value) {
is CallState.IncomingCall -> {
// Best-effort on a detached scope. If the process
// dies before completion, the remote 60s timeout
// or our 65s watchdog handles it.
kotlinx.coroutines.CoroutineScope(
kotlinx.coroutines.SupervisorJob() + kotlinx.coroutines.Dispatchers.Main.immediate,
).launch {
manager.rejectCall()
}
}
is CallState.Offering,
is CallState.Connecting,
is CallState.Connected,
-> {
kotlinx.coroutines.CoroutineScope(
kotlinx.coroutines.SupervisorJob() + kotlinx.coroutines.Dispatchers.Main.immediate,
).launch {
manager.hangup()
}
}
else -> {}
}
}
super.onDestroy()
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt
+32 -20
View File
@@ -86,10 +86,13 @@ class CallManager(
* missing buffered events.
* - No null-check on every emit site.
*
* Buffer size 16 covers bursts of ICE candidates + answers arriving
* in quick succession during call setup.
* Buffer size 256 handles worst-case bursts (20+ ICE candidates per
* peer × multiple peers + answers + peer-left events). tryEmit is
* used instead of emit to avoid suspending while holding stateMutex
* (which would block all signaling). Drops are logged but should
* never happen in practice with this buffer size.
*/
private val _sessionEvents = MutableSharedFlow<CallSessionEvent>(extraBufferCapacity = 16)
private val _sessionEvents = MutableSharedFlow<CallSessionEvent>(extraBufferCapacity = 256)
val sessionEvents: SharedFlow<CallSessionEvent> = _sessionEvents.asSharedFlow()
/**
@@ -97,7 +100,14 @@ class CallManager(
* because renegotiation has its own glare-resolution logic in
* CallSession and benefits from a dedicated collector.
*/
private val _renegotiationEvents = MutableSharedFlow<CallRenegotiateEvent>(extraBufferCapacity = 8)
/** Emits a session event, logging a warning if the buffer overflows. */
private fun emitSessionEvent(event: CallSessionEvent) {
if (!emitSessionEvent(event)) {
Log.e("CallManager") { "sessionEvents buffer overflow — dropped: $event" }
}
}
private val _renegotiationEvents = MutableSharedFlow<CallRenegotiateEvent>(extraBufferCapacity = 32)
val renegotiationEvents: SharedFlow<CallRenegotiateEvent> = _renegotiationEvents.asSharedFlow()
private var timeoutJob: Job? = null
@@ -350,7 +360,7 @@ class CallManager(
else -> {}
}
_sessionEvents.tryEmit(CallSessionEvent.MidCallOfferReceived(callerPubKey, event.sdpOffer()))
emitSessionEvent(CallSessionEvent.MidCallOfferReceived(callerPubKey, event.sdpOffer()))
return
}
}
@@ -451,7 +461,7 @@ class CallManager(
if (discovered.isNotEmpty()) {
Log.d("CallManager") { "acceptCall: triggering mesh setup with ${discovered.size} discovered peers: ${discovered.map { it.take(8) }}" }
for (peer in discovered) {
_sessionEvents.tryEmit(CallSessionEvent.NewPeerInGroupCall(peer))
emitSessionEvent(CallSessionEvent.NewPeerInGroupCall(peer))
}
}
}
@@ -518,7 +528,7 @@ class CallManager(
disarmRingingWatchdog()
armConnectingWatchdog(current.callId)
Log.d("CallManager") { "onCallAnswered: Offering -> Connecting, forwarding answer to CallSession" }
_sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event))
emitSessionEvent(CallSessionEvent.AnswerReceived(event))
}
is CallState.Connecting -> {
@@ -547,7 +557,7 @@ class CallManager(
// Forward to CallController — it routes to the correct PeerSession
// and internally triggers callee-to-callee mesh setup if needed.
Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding to CallController" }
_sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event))
emitSessionEvent(CallSessionEvent.AnswerReceived(event))
}
is CallState.IncomingCall -> {
@@ -586,7 +596,7 @@ class CallManager(
// 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" }
_sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event))
emitSessionEvent(CallSessionEvent.AnswerReceived(event))
}
else -> {
@@ -626,7 +636,7 @@ class CallManager(
transitionToEnded(current.callId, current.peerPubKeys, EndReason.PEER_REJECTED)
} else {
_state.value = current.copy(peerPubKeys = remaining)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer))
emitSessionEvent(CallSessionEvent.PeerLeft(rejectingPeer))
}
}
@@ -635,7 +645,7 @@ class CallManager(
cancelPeerTimeout(rejectingPeer)
_state.value =
current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer))
emitSessionEvent(CallSessionEvent.PeerLeft(rejectingPeer))
}
is CallState.Connected -> {
@@ -643,7 +653,7 @@ class CallManager(
cancelPeerTimeout(rejectingPeer)
_state.value =
current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer))
emitSessionEvent(CallSessionEvent.PeerLeft(rejectingPeer))
}
is CallState.IncomingCall -> {
@@ -670,7 +680,7 @@ class CallManager(
}
private fun onIceCandidate(event: CallIceCandidateEvent) {
_sessionEvents.tryEmit(CallSessionEvent.IceCandidateReceived(event))
emitSessionEvent(CallSessionEvent.IceCandidateReceived(event))
}
private fun onRenegotiate(event: CallRenegotiateEvent) {
@@ -683,7 +693,9 @@ class CallManager(
else -> return
}
if (callId != currentCallId) return
_renegotiationEvents.tryEmit(event)
if (!_renegotiationEvents.tryEmit(event)) {
Log.e("CallManager") { "renegotiationEvents buffer overflow — dropped renegotiation from ${event.pubKey.take(8)}" }
}
}
suspend fun sendRenegotiation(
@@ -832,7 +844,7 @@ class CallManager(
peerPubKeys = connectedRemaining,
pendingPeerPubKeys = pendingRemaining,
)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer))
emitSessionEvent(CallSessionEvent.PeerLeft(leavingPeer))
}
}
@@ -857,7 +869,7 @@ class CallManager(
peerPubKeys = connectedRemaining,
pendingPeerPubKeys = pendingRemaining,
)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer))
emitSessionEvent(CallSessionEvent.PeerLeft(leavingPeer))
}
}
@@ -869,7 +881,7 @@ class CallManager(
transitionToEnded(callId, current.peerPubKeys, EndReason.PEER_HANGUP)
} else {
_state.value = current.copy(peerPubKeys = remaining)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer))
emitSessionEvent(CallSessionEvent.PeerLeft(leavingPeer))
}
}
@@ -1187,7 +1199,7 @@ class CallManager(
transitionToEnded(current.callId, current.peerPubKeys, EndReason.TIMEOUT)
} else {
_state.value = current.copy(peerPubKeys = remaining)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey))
emitSessionEvent(CallSessionEvent.PeerLeft(peerPubKey))
}
}
@@ -1205,7 +1217,7 @@ class CallManager(
)
} else {
_state.value = current.copy(pendingPeerPubKeys = newPending)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey))
emitSessionEvent(CallSessionEvent.PeerLeft(peerPubKey))
}
}
@@ -1215,7 +1227,7 @@ class CallManager(
Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Connected (invitedByUs=$wasInvitedByUs)" }
shouldPublishHangup = wasInvitedByUs
_state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - peerPubKey)
_sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey))
emitSessionEvent(CallSessionEvent.PeerLeft(peerPubKey))
}
else -> {