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

fix(audio-rooms): round-2 audit — pump self-join (CRITICAL) + 5 other findings

Browse Source 
Round 2 audit (3 parallel agents reviewing every change since the
previous follow-up commit) caught one CRITICAL regression and several
HIGH/MED items. Most round-1 fixes verified clean.

CRITICAL fix (audit round-2 MoQ #1):
- Pump exception handlers added in the previous commit call `close()`
  from inside the failing pump's own coroutine. `close()` now does
  `controlPumpJob?.join()` to drain in-flight writes — but the Job we
  try to join is the very Job we're inside, so `join()` suspends
  forever (lambda can't finish until close returns; close can't
  return until lambda finishes). `runCatching` doesn't help — `join()`
  doesn't throw, it suspends. This deadlocks the entire session
  whenever a pump fails.
  Fix: skip the join when the current coroutine IS the job we're
  joining. `currentCoroutineContext()[Job]` identifies the caller; we
  compare and bypass.

HIGH fixes:
- MoQ #2 (regression): concurrent unannounce() + post-OK
  AnnounceError handler could both write UNANNOUNCE on the wire (some
  relays disconnect on UNANNOUNCE for an unknown namespace). Fix:
  `AnnounceHandleImpl.unannounceWritten: Boolean` flag, set under
  stateMutex by whichever writer goes first; the other path skips.
- VM #3 (new): `connect()` overwrote `listener` if invoked from a
  Failed-with-stale-listener state, leaking the previous MoQ session.
  Fix: call `teardown(targetState=Idle, finalCleanup=false)` before
  launching the new attempt when listener or stateObserverJob is
  still alive.
- VM #7 (new): `openSubscription` allocated decoder + player via the
  factories, then attached them to the slot. If the VM scope was
  cancelled between `decoderFactory()` and `slot.attach(...)`, the
  native MediaCodec / AudioTrack leaked because nothing was tracking
  them yet. Fix: nest a try/catch that releases both on any throw
  (including `CancellationException`) before re-throwing.

MED fixes:
- MoQ #7 (new): `capture.start()` could throw before `job` was
  assigned, leaving the broadcaster in a half-started state where
  future `start()` calls would re-pass the guards and double-start
  the mic. Fix: try/catch around capture.start; on throw, set
  `stopped = true` + run capture.stop and propagate.
- MoQ #8 (new): `stopped` was read across threads (setMuted from
  caller, stop from anywhere) without a `@Volatile` barrier.
  Visibility hazard. Fix: `@Volatile private var stopped`.
- Android #12 (new): after the user granted RECORD_AUDIO via the
  Settings deep-link, `permissionDenied` stayed `true` because the
  launcher callback never fired — the warning + Open-settings button
  remained visible until the user tapped Talk again. Fix: derive
  `showDenialWarning` from `permissionDenied AND
  ContextCompat.checkSelfPermission(...) != GRANTED`. Re-checks every
  recomposition (including post-Settings return).

Round-1 fixes verified clean by this audit:
- pending-deferred completeExceptionally on close
- SubscribeDoneStatus codes (UNSUBSCRIBED=0x00, TRACK_ENDED=0x03)
- suspend `stop()` conversions on broadcaster + player
- gate release before NestsSpeaker teardown chain
- unannounce() ordering on thrown wire-write
- SharedFlow PIP signal (rapid double-tap behavior is correct)
- RECEIVER_NOT_EXPORTED gate (constant 4 doesn't collide; round-1
  collision claim was incorrect)
- onUserLeaveHint guards (PIP from lobby, no PIP support)
- foreground service `startForeground`-always-first contract
  (`Result.onFailure` is `inline`, the `return` IS a non-local
  return from `onStartCommand` — verified)
- 4-hour wake-lock cap
- AudioRoomBridge.clear() in AccountViewModel.onCleared

Still deferred:
- VM #6 (round-1 carryover): unsynchronized speakingExpiryJobs map.
  Cross-thread mutation under contention. Needs ConcurrentHashMap or
  Dispatchers.Main.immediate marshalling.
- VM #10 (round-2 new): brief two-QUIC-session overlap during
  rapid disconnect→connect.
- Android #5 (round-2 new): same-room re-entry from MainActivity
  while in PIP doesn't auto-exit PIP.
- MoQ #1, #8 (round-1 carryover): wire-format draft pinning. Still
  blocked on M4 manual interop input.

Verified: spotlessApply clean; :commons:jvmTest, :nestsClient:jvmTest
(80 tests), :amethyst:compilePlayDebugKotlin all green.
This commit is contained in:
Claude
2026-04-26 12:32:22 +00:00
parent 0b1ec52f79
commit ed793e8eb3
4 changed files with 123 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomFullScreen.kt
+16 -2
View File
@@ -228,8 +228,22 @@ private fun TalkRow(
var permissionDenied by rememberSaveable { mutableStateOf(false) }
val permissionLauncher =
rememberLauncherForActivityResult(ActivityResultContracts.RequestPermission()) { granted ->
if (granted) viewModel.startBroadcast(speakerPubkeyHex) else permissionDenied = true
if (granted) {
permissionDenied = false
viewModel.startBroadcast(speakerPubkeyHex)
} else {
permissionDenied = true
}
}
// If the user grants RECORD_AUDIO via the system Settings deep-link
// and returns to the activity, the permissionLauncher callback never
// fires and `permissionDenied` would otherwise stay true. Recompute
// every time the permission state could have changed (audit round-2
// Android #12).
val showDenialWarning =
permissionDenied &&
ContextCompat.checkSelfPermission(context, Manifest.permission.RECORD_AUDIO) !=
PackageManager.PERMISSION_GRANTED
Row(
modifier = Modifier.fillMaxWidth().padding(top = 8.dp),
@@ -252,7 +266,7 @@ private fun TalkRow(
}) {
Text(stringRes(R.string.audio_room_talk))
}
if (permissionDenied) {
if (showDenialWarning) {
Text(
text = stringRes(R.string.audio_room_record_permission_required),
style = MaterialTheme.typography.bodySmall,
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt
+42 -17
View File
@@ -164,11 +164,16 @@ class AudioRoomViewModel(
autoRetryJob = null
autoRetryAttempts = 0
retryPending = false
// Cancel any previous state observer too — otherwise its delayed
// emissions can clobber the fresh Connecting UI we set below
// (audit VM #8).
stateObserverJob?.cancel()
stateObserverJob = null
// If a stale listener is still set (e.g. arrived in Failed and the
// user is manually retrying before the auto-retry fires, or
// entering from a server-Closed-but-not-yet-disconnected state),
// tear it down so the new connect doesn't leave the old MoQ
// session open and unowned (audit round-2 VM #3). teardown()
// also cancels stateObserverJob so its delayed emissions can't
// clobber the fresh Connecting UI.
if (listener != null || stateObserverJob != null) {
teardown(targetState = ConnectionUiState.Idle, finalCleanup = false)
}
_uiState.update { it.copy(connection = ConnectionUiState.Connecting(ConnectionUiState.Step.ResolvingRoom)) }
@@ -549,19 +554,39 @@ class AudioRoomViewModel(
viewModelScope.launch { runCatching { handle.unsubscribe() } }
return
}
// Allocate native resources (MediaCodec decoder + AudioTrack
// player on Android). Both are heavy and leaky if dropped on
// the floor — wrap them in a nested try so any cancellation
// or throw between here and slot.attach releases them
// (audit round-2 VM #7).
val decoder = decoderFactory()
val player = playerFactory()
val isMuted = _uiState.value.isMuted
val roomPlayer = AudioRoomPlayer(decoder, player, viewModelScope)
// Apply current mute state before play() opens the device so the
// first frame respects it.
player.setMutedSafe(isMuted)
// Tap the object flow to drive the speaking-now indicator before
// the decoder consumes it.
val instrumented = handle.objects.onEach { onSpeakerActivity(pubkey) }
roomPlayer.play(instrumented, onError = { /* swallow per-packet decoder errors */ })
slot.attach(handle, roomPlayer, player)
publishActiveSpeakers()
val player =
try {
playerFactory()
} catch (t: Throwable) {
runCatching { decoder.release() }
throw t
}
try {
val isMuted = _uiState.value.isMuted
val roomPlayer = AudioRoomPlayer(decoder, player, viewModelScope)
// Apply current mute state before play() opens the device so the
// first frame respects it.
player.setMutedSafe(isMuted)
// Tap the object flow to drive the speaking-now indicator before
// the decoder consumes it.
val instrumented = handle.objects.onEach { onSpeakerActivity(pubkey) }
roomPlayer.play(instrumented, onError = { /* swallow per-packet decoder errors */ })
slot.attach(handle, roomPlayer, player)
publishActiveSpeakers()
} catch (t: Throwable) {
// Either CancellationException (scope cancelled mid-construction)
// or an unexpected throw — release the half-built pipeline and
// re-throw so the outer catch handles slot rollback.
runCatching { player.stop() }
runCatching { decoder.release() }
throw t
}
} catch (ce: CancellationException) {
throw ce
} catch (t: Throwable) {
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt
+17 -3
View File
@@ -53,19 +53,33 @@ class AudioRoomBroadcaster(
private val scope: CoroutineScope,
) {
private var job: Job? = null
private var stopped = false
@Volatile private var stopped: Boolean = false
@Volatile private var muted: Boolean = false
/**
* Start capturing + encoding + publishing in the background. Returns
* immediately. Calling twice is an error.
* immediately. Calling twice is an error. If [AudioCapture.start]
* throws (mic device unavailable, etc.), the broadcaster is left in
* a stopped state and the exception propagates so the caller can
* surface it to the user.
*/
fun start(onError: (AudioException) -> Unit = { /* swallow */ }) {
check(!stopped) { "AudioRoomBroadcaster already stopped" }
check(job == null) { "AudioRoomBroadcaster.start already called" }
capture.start()
// Audit round-2 MoQ #7: capture.start() can throw before we have
// a job. Mark stopped + propagate so a half-started capture isn't
// left holding the mic and the broadcaster instance can't be
// accidentally re-started.
try {
capture.start()
} catch (t: Throwable) {
stopped = true
runCatching { capture.stop() }
throw t
}
job =
scope.launch {
try {
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt
+48 -8
View File
@@ -30,6 +30,7 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.receiveAsFlow
@@ -472,9 +473,23 @@ class MoqSession private constructor(
}
}
handle?.let { existing ->
stateMutex.withLock { existing.markSessionClosedLocked() }
runCatching {
writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(msg.namespace))) }
val shouldWrite =
stateMutex.withLock {
existing.markSessionClosedLocked()
// Same compare-and-set as unannounce() to avoid
// a duplicate UNANNOUNCE on the wire when a
// concurrent unannounce() is mid-flight.
if (existing.unannounceWritten) {
false
} else {
existing.unannounceWritten = true
true
}
}
if (shouldWrite) {
runCatching {
writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(msg.namespace))) }
}
}
stateMutex.withLock { announces.remove(msg.namespace) }
}
@@ -621,14 +636,21 @@ class MoqSession private constructor(
// (handleInboundSubscribe writing SUBSCRIBE_OK / SUBSCRIBE_ERROR,
// handleInboundUnsubscribe writing SUBSCRIBE_DONE) finishes its
// writeMutex.withLock { ... } critical section before we send FIN
// on the control stream. Joining a cancelled coroutine waits for
// its `finally`s to run, including the lock release.
// on the control stream.
//
// Self-join guard: when close() is called from a pump's own
// exception handler (the try/catch in startPumps), `coroutineContext[Job]`
// is the very Job we're trying to join. Joining ourselves
// suspends forever — the lambda can't finish until close() returns,
// and close() can't return until the lambda finishes. Skip the
// join in that case (round-2 audit MoQ #1).
val ctrlPump = controlPumpJob
val datagramPump = datagramPumpJob
ctrlPump?.cancel()
datagramPump?.cancel()
runCatching { ctrlPump?.join() }
runCatching { datagramPump?.join() }
val self = currentCoroutineContext()[Job]
runCatching { if (ctrlPump != null && ctrlPump !== self) ctrlPump.join() }
runCatching { if (datagramPump != null && datagramPump !== self) datagramPump.join() }
runCatching { writeMutex.withLock { controlStream.finish() } }
runCatching { transport.close(code, reason) }
}
@@ -704,6 +726,11 @@ class MoqSession private constructor(
internal val tracks = HashMap<TrackKey, TrackPublisherImpl>()
internal var sessionClosed = false
// True once UNANNOUNCE has been written on the wire, so a
// concurrent unannounce() + post-OK AnnounceError handler don't
// both emit it (audit round-2 MoQ #2). Read/written under stateMutex.
internal var unannounceWritten = false
override suspend fun openTrack(name: ByteArray): TrackPublisher {
val key = TrackKey(name)
val publisher = TrackPublisherImpl(name)
@@ -736,7 +763,20 @@ class MoqSession private constructor(
// thinks the namespace exists but our state has dropped it.
}
toClose.forEach { runCatching { it.close() } }
if (!closed) {
// Race guard: a post-OK AnnounceError handler may have already
// written UNANNOUNCE for this namespace (audit round-2 MoQ #2).
// Compare-and-set under stateMutex so only one writer fires
// the wire frame.
val shouldWrite =
stateMutex.withLock {
if (unannounceWritten) {
false
} else {
unannounceWritten = true
true
}
}
if (shouldWrite && !closed) {
runCatching {
writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(namespace))) }
}