fix: WebRTC call resource leaks, thread safety, and error handling

- Send hangup/reject to peer when WebRTC init or PeerConnection creation
  fails, so remote phone stops ringing instead of timing out after 60s
- Throw on null PeerConnection from factory to fail fast instead of
  silently no-oping all subsequent WebRTC operations
- Start foreground service during IncomingCall to protect ringtone
  playback from being killed on Android 14+
- Make cleanup() idempotent with AtomicBoolean guard to prevent double
  disposal when Ended state and ViewModel.onCleared race
- Replace mutableMapOf with ConcurrentHashMap for videoSenders accessed
  from UI and WebRTC callback threads
- Add @Volatile to peerConnection, videoPausedByProximity, and
  foregroundServiceStarted for cross-thread visibility
- Capture peerConnection into local variable in dispose() to prevent
  TOCTOU race between close() and null assignment
- Replace leaked MainScope() in CallNotificationReceiver with structured
  CoroutineScope that is cancelled after work completes
- Remove self-wraps in group answer/reject to avoid wasting bandwidth
  sending encrypted messages to ourselves
- Move startTimeout inside stateMutex in initiateCall for consistency

https://claude.ai/code/session_017HrFJNxD6zrGwiZ3s69xTh
This commit is contained in:
Claude
2026-04-10 23:55:16 +00:00
parent 6c28f9d995
commit d851168a46
4 changed files with 35 additions and 16 deletions
@@ -53,6 +53,8 @@ import org.webrtc.EglBase
import org.webrtc.IceCandidate import org.webrtc.IceCandidate
import org.webrtc.VideoTrack import org.webrtc.VideoTrack
import java.util.UUID import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicBoolean
private const val TAG = "CallController" private const val TAG = "CallController"
private const val VIDEO_MAX_BITRATE_BPS_DEFAULT = 1_500_000 private const val VIDEO_MAX_BITRATE_BPS_DEFAULT = 1_500_000
@@ -95,9 +97,11 @@ class CallController(
val audioRoute: StateFlow<AudioRoute> = audioManager.audioRoute val audioRoute: StateFlow<AudioRoute> = audioManager.audioRoute
val isBluetoothAvailable: StateFlow<Boolean> = audioManager.isBluetoothAvailable val isBluetoothAvailable: StateFlow<Boolean> = audioManager.isBluetoothAvailable
private var videoPausedByProximity = false @Volatile private var videoPausedByProximity = false
private var foregroundServiceStarted = false
private val videoSenders = mutableMapOf<HexKey, org.webrtc.RtpSender>() @Volatile private var foregroundServiceStarted = false
private val cleanedUp = AtomicBoolean(false)
private val videoSenders = ConcurrentHashMap<HexKey, org.webrtc.RtpSender>()
private val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager private val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
private var networkCallbackRegistered = false private var networkCallbackRegistered = false
@@ -126,6 +130,7 @@ class CallController(
callManager.state.collect { state -> callManager.state.collect { state ->
when (state) { when (state) {
is CallState.IncomingCall -> { is CallState.IncomingCall -> {
ensureForegroundService()
withContext(Dispatchers.IO) { audioManager.startRinging() } withContext(Dispatchers.IO) { audioManager.startRinging() }
scope.launch { scope.launch {
NotificationUtils.showIncomingCallNotification(state.callerPubKey, context) NotificationUtils.showIncomingCallNotification(state.callerPubKey, context)
@@ -209,6 +214,7 @@ class CallController(
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Failed to initialize WebRTC", e) Log.e(TAG, "Failed to initialize WebRTC", e)
_errorMessage.value = "Failed to start call: ${e.message}" _errorMessage.value = "Failed to start call: ${e.message}"
callManager.hangup()
return@launch return@launch
} }
@@ -247,6 +253,7 @@ class CallController(
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Failed to initialize WebRTC", e) Log.e(TAG, "Failed to initialize WebRTC", e)
_errorMessage.value = "Failed to accept call: ${e.message}" _errorMessage.value = "Failed to accept call: ${e.message}"
callManager.rejectCall()
return@launch return@launch
} }
@@ -256,6 +263,7 @@ class CallController(
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Failed to create PeerConnection", e) Log.e(TAG, "Failed to create PeerConnection", e)
_errorMessage.value = "Failed to accept call: ${e.message}" _errorMessage.value = "Failed to accept call: ${e.message}"
callManager.rejectCall()
return@launch return@launch
} }
@@ -582,6 +590,7 @@ class CallController(
// ---- Cleanup ---- // ---- Cleanup ----
fun cleanup() { fun cleanup() {
if (!cleanedUp.compareAndSet(false, true)) return
unregisterNetworkCallback() unregisterNetworkCallback()
try { try {
audioManager.release() audioManager.release()
@@ -610,6 +619,7 @@ class CallController(
_isAudioMuted.value = false _isAudioMuted.value = false
videoPausedByProximity = false videoPausedByProximity = false
videoSenders.clear() videoSenders.clear()
cleanedUp.set(false)
} }
// ---- Foreground service ---- // ---- Foreground service ----
@@ -24,6 +24,9 @@ import android.content.BroadcastReceiver
import android.content.Context import android.content.Context
import android.content.Intent import android.content.Intent
import com.vitorpamplona.amethyst.service.notifications.NotificationUtils import com.vitorpamplona.amethyst.service.notifications.NotificationUtils
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
/** /**
@@ -40,7 +43,8 @@ class CallNotificationReceiver : BroadcastReceiver() {
) { ) {
val callManager = CallSessionBridge.callManager ?: return val callManager = CallSessionBridge.callManager ?: return
val pendingResult = goAsync() val pendingResult = goAsync()
kotlinx.coroutines.MainScope().launch { val scope = CoroutineScope(SupervisorJob() + kotlinx.coroutines.Dispatchers.Main.immediate)
scope.launch {
try { try {
when (intent.action) { when (intent.action) {
ACTION_REJECT_CALL -> { ACTION_REJECT_CALL -> {
@@ -54,6 +58,7 @@ class CallNotificationReceiver : BroadcastReceiver() {
} }
} finally { } finally {
pendingResult.finish() pendingResult.finish()
scope.cancel()
} }
} }
} }
@@ -55,8 +55,9 @@ class WebRtcCallSession(
private val onRenegotiationNeeded: () -> Unit = {}, private val onRenegotiationNeeded: () -> Unit = {},
private val onIceRestartOffer: (SessionDescription) -> Unit = {}, private val onIceRestartOffer: (SessionDescription) -> Unit = {},
) { ) {
private var peerConnection: PeerConnection? = null @Volatile private var peerConnection: PeerConnection? = null
private var iceRestartAttempted = false
@Volatile private var iceRestartAttempted = false
fun createPeerConnection() { fun createPeerConnection() {
val rtcConfig = val rtcConfig =
@@ -65,7 +66,7 @@ class WebRtcCallSession(
continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY
} }
peerConnection = val pc =
peerConnectionFactory.createPeerConnection( peerConnectionFactory.createPeerConnection(
rtcConfig, rtcConfig,
object : PeerConnection.Observer { object : PeerConnection.Observer {
@@ -155,7 +156,8 @@ class WebRtcCallSession(
} }
} }
}, },
) ) ?: throw IllegalStateException("PeerConnectionFactory.createPeerConnection returned null")
peerConnection = pc
} }
/** /**
@@ -345,9 +347,10 @@ class WebRtcCallSession(
} }
fun dispose() { fun dispose() {
peerConnection?.close() val pc = peerConnection ?: return
peerConnection?.dispose()
peerConnection = null peerConnection = null
pc.close()
pc.dispose()
} }
private fun loggingSdpObserver(label: String) = private fun loggingSdpObserver(label: String) =
@@ -201,9 +201,9 @@ class CallManager(
val result = factory.createCallOffer(sdpOffer, calleePubKey, callId, callType, signer) val result = factory.createCallOffer(sdpOffer, calleePubKey, callId, callType, signer)
stateMutex.withLock { stateMutex.withLock {
_state.value = CallState.Offering(callId, setOf(calleePubKey), callType) _state.value = CallState.Offering(callId, setOf(calleePubKey), callType)
startTimeout(callId)
} }
publishEvent(result.wrap) publishEvent(result.wrap)
startTimeout(callId)
Log.d("CallManager") { "initiateCall: offer published, timeout started" } Log.d("CallManager") { "initiateCall: offer published, timeout started" }
} }
@@ -281,9 +281,10 @@ class CallManager(
discoveredCalleePeers.clear() discoveredCalleePeers.clear()
} }
val allRecipients = current.groupMembers + signer.pubKey val allMembers = current.groupMembers + signer.pubKey
Log.d("CallManager") { "acceptCall: publishing answer to ${allRecipients.size} recipients" } val otherMembers = allMembers - signer.pubKey
val result = factory.createGroupCallAnswer(sdpAnswer, allRecipients, current.callId, signer) Log.d("CallManager") { "acceptCall: publishing answer to ${otherMembers.size} recipients" }
val result = factory.createGroupCallAnswer(sdpAnswer, otherMembers, current.callId, signer)
result.wraps.forEach { publishEvent(it) } result.wraps.forEach { publishEvent(it) }
Log.d("CallManager") { "acceptCall: answer published, now in Connecting state" } Log.d("CallManager") { "acceptCall: answer published, now in Connecting state" }
@@ -306,8 +307,8 @@ class CallManager(
transitionToEnded(current.callId, current.peerPubKeys(), EndReason.REJECTED) transitionToEnded(current.callId, current.peerPubKeys(), EndReason.REJECTED)
} }
val allRecipients = current.groupMembers + signer.pubKey val otherMembers = current.groupMembers - signer.pubKey
val result = factory.createGroupReject(allRecipients, current.callId, signer = signer) val result = factory.createGroupReject(otherMembers, current.callId, signer = signer)
result.wraps.forEach { publishEvent(it) } result.wraps.forEach { publishEvent(it) }
} }