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

fix: WebRTC race conditions, thread safety, and PiP lifecycle

Browse Source 
- CallActivity.onStop: move finishAndRemoveTask inside coroutine so
  hangup signaling completes before Activity destruction; add
  hangupInitiated flag to prevent double-hangup in onDestroy
- CallController: add per-peer renegotiation debouncing via
  pendingRenegotiation map to prevent queuing multiple createOffer calls
  when video is toggled rapidly
- CallController: guard ensureForegroundService in onPeerConnected
  callback with state check to prevent restarting the service after
  cleanup
- RemoteVideoMonitor: synchronize onRemoteVideoTrack, onPeerRemoved,
  and dispose with trackLock to prevent non-atomic map mutations and
  leaked video sinks from concurrent WebRTC callback threads
- CallMediaManager: add @Synchronized to createVideoResources to
  prevent check-then-act race between IO and main threads

https://claude.ai/code/session_017HrFJNxD6zrGwiZ3s69xTh
This commit is contained in:
Claude
2026-04-11 00:48:27 +00:00
parent deecce6e77
commit c82ba766f8
4 changed files with 76 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallController.kt
+17 -3
View File
@@ -103,6 +103,8 @@ class CallController(
private val cleanedUp = AtomicBoolean(false)
private val videoSenders = ConcurrentHashMap<HexKey, org.webrtc.RtpSender>()
private val pendingRenegotiation = ConcurrentHashMap<HexKey, Boolean>()
private val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
private var networkCallbackRegistered = false
private val networkCallback =
@@ -409,10 +411,19 @@ class CallController(
}
private fun performRenegotiation(peerPubKey: HexKey) {
val webRtcSession = webRtcSession(peerPubKey) ?: return
if (pendingRenegotiation.putIfAbsent(peerPubKey, true) != null) return
val webRtcSession =
webRtcSession(peerPubKey) ?: run {
pendingRenegotiation.remove(peerPubKey)
return
}
val state = callManager.state.value
if (state !is CallState.Connected && state !is CallState.Connecting) return
if (state !is CallState.Connected && state !is CallState.Connecting) {
pendingRenegotiation.remove(peerPubKey)
return
}
webRtcSession.createOffer { sdp ->
pendingRenegotiation.remove(peerPubKey)
scope.launch { callManager.sendRenegotiation(sdp.description, peerPubKey) }
}
}
@@ -540,7 +551,9 @@ class CallController(
Log.d(TAG) { "Peer ${peerPubKey.take(8)} connected!" }
scope.launch {
callManager.onPeerConnected()
ensureForegroundService()
if (callManager.state.value is CallState.Connected) {
ensureForegroundService()
}
}
},
onRemoteVideoTrack = { track -> videoMonitor.onRemoteVideoTrack(peerPubKey, track) },
@@ -627,6 +640,7 @@ class CallController(
_isAudioMuted.value = false
videoPausedByProximity = false
videoSenders.clear()
pendingRenegotiation.clear()
cleanedUp.set(false)
}
amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallMediaManager.kt
+1
View File
@@ -117,6 +117,7 @@ class CallMediaManager(
}
}
@Synchronized
fun createVideoResources() {
if (localVideoSource != null) return
val factory = peerConnectionFactory ?: return
amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/RemoteVideoMonitor.kt
+32 -22
View File
@@ -85,46 +85,56 @@ class RemoteVideoMonitor(
private val perPeerLastFrameTimeMs = ConcurrentHashMap<HexKey, AtomicLong>()
private var groupVideoMonitorJob: Job? = null
/** Protects compound read-modify-write on [_remoteVideoTracks] and
* [_remoteVideoTrack] which can be called from WebRTC callback threads. */
private val trackLock = Any()
fun onRemoteVideoTrack(
peerPubKey: HexKey,
track: VideoTrack,
) {
Log.d(TAG) { "Remote video track from ${peerPubKey.take(8)}" }
_remoteVideoTracks.value = _remoteVideoTracks.value + (peerPubKey to track)
if (_remoteVideoTrack.value == null) {
_remoteVideoTrack.value = track
startPrimaryMonitor(track)
synchronized(trackLock) {
_remoteVideoTracks.value = _remoteVideoTracks.value + (peerPubKey to track)
if (_remoteVideoTrack.value == null) {
_remoteVideoTrack.value = track
startPrimaryMonitor(track)
}
}
startPeerMonitor(peerPubKey, track)
}
fun onPeerRemoved(peerPubKey: HexKey) {
stopPeerMonitor(peerPubKey)
val currentTracks = _remoteVideoTracks.value
if (peerPubKey in currentTracks) {
_remoteVideoTracks.value = currentTracks - peerPubKey
if (_remoteVideoTrack.value == currentTracks[peerPubKey]) {
stopPrimaryMonitor()
val nextTrack = _remoteVideoTracks.value.values.firstOrNull()
_remoteVideoTrack.value = nextTrack
if (nextTrack != null) {
startPrimaryMonitor(nextTrack)
synchronized(trackLock) {
val currentTracks = _remoteVideoTracks.value
if (peerPubKey in currentTracks) {
_remoteVideoTracks.value = currentTracks - peerPubKey
if (_remoteVideoTrack.value == currentTracks[peerPubKey]) {
stopPrimaryMonitor()
val nextTrack = _remoteVideoTracks.value.values.firstOrNull()
_remoteVideoTrack.value = nextTrack
if (nextTrack != null) {
startPrimaryMonitor(nextTrack)
}
}
}
}
}
fun dispose() {
stopPrimaryMonitor()
stopGroupMonitor()
for (peerPubKey in perPeerFrameSinks.keys.toList()) {
stopPeerMonitor(peerPubKey)
synchronized(trackLock) {
stopPrimaryMonitor()
stopGroupMonitor()
for (peerPubKey in perPeerFrameSinks.keys.toList()) {
stopPeerMonitor(peerPubKey)
}
_remoteVideoTrack.value = null
_remoteVideoTracks.value = emptyMap()
_isRemoteVideoActive.value = false
_remoteVideoAspectRatio.value = null
_activePeerVideos.value = emptySet()
}
_remoteVideoTrack.value = null
_remoteVideoTracks.value = emptyMap()
_isRemoteVideoActive.value = false
_remoteVideoAspectRatio.value = null
_activePeerVideos.value = emptySet()
}
private fun startPrimaryMonitor(track: VideoTrack) {
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/CallActivity.kt
+26 -22
View File
@@ -48,10 +48,8 @@ import com.vitorpamplona.amethyst.ui.screen.ManageWebOkHttp
import com.vitorpamplona.amethyst.ui.theme.AmethystTheme
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
class CallActivity : AppCompatActivity() {
val isInPipMode = mutableStateOf(false)
@@ -69,6 +67,7 @@ class CallActivity : AppCompatActivity() {
}
private var pendingAcceptIsVideo = false
private var hangupInitiated = false
private val pipActionReceiver =
object : BroadcastReceiver() {
@@ -201,13 +200,17 @@ class CallActivity : AppCompatActivity() {
// We must NOT hang up when the user simply presses Home from the full-screen
// call UI (that enters PiP via onUserLeaveHint instead).
if (wasInPipMode && !isInPictureInPictureMode) {
val state = CallSessionBridge.callManager?.state?.value
hangupInitiated = true
val manager = CallSessionBridge.callManager
val state = manager?.state?.value
if (state is CallState.Connected || state is CallState.Connecting || state is CallState.Offering) {
lifecycleScope.launch {
withContext(NonCancellable) { CallSessionBridge.callManager?.hangup() }
CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate).launch {
manager.hangup()
finishAndRemoveTask()
}
} else {
finishAndRemoveTask()
}
finishAndRemoveTask()
}
}
@@ -216,26 +219,27 @@ class CallActivity : AppCompatActivity() {
// Safety net: if the Activity is destroyed while a call is still
// ringing/offering, ensure the call is hung up so audio stops.
// Use a standalone CoroutineScope because lifecycleScope is cancelled
// during super.onDestroy() and may drop suspend work.
val manager = CallSessionBridge.callManager
when (manager?.state?.value) {
is CallState.IncomingCall -> {
CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate).launch {
manager.rejectCall()
// Skip if onStop already initiated the hangup to avoid double signaling.
if (!hangupInitiated) {
val manager = CallSessionBridge.callManager
when (manager?.state?.value) {
is CallState.IncomingCall -> {
CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate).launch {
manager.rejectCall()
}
}
}
is CallState.Offering,
is CallState.Connecting,
is CallState.Connected,
-> {
CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate).launch {
manager.hangup()
is CallState.Offering,
is CallState.Connecting,
is CallState.Connected,
-> {
CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate).launch {
manager.hangup()
}
}
}
else -> {}
else -> {}
}
}
super.onDestroy()