fix(audio-rooms): audit follow-up — MoQ HIGH/MED + VM concurrency + Android polish
Round 2 of the audit-driven cleanup. Lands every HIGH and most MED findings from the protocol / ViewModel / Android lifecycle audits that weren't fixed in the previous commit. The wire-format draft pinning (audit MoQ #1, #8) is intentionally deferred until the M4 manual interop pass against `nostrnests.com` reveals which draft revision the relay is actually speaking. MoQ session fixes: - #5 UNANNOUNCE wire-write now happens BEFORE removing announces[ns], so an inbound SUBSCRIBE during the teardown window sees the namespace as withdrawn (sessionClosed=true → SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST)) instead of "namespace never existed". - #4 dispatchControlMessage(AnnounceError) now distinguishes pre-OK and post-OK errors. Post-OK is a session-level kick: mark the handle closed, send UNANNOUNCE, then drop. Pre-OK still rolls back the optimistic announces[] insert as before. - #6 close() now joins the cancelled control + datagram pumps before calling controlStream.finish(), so an in-flight SUBSCRIBE_OK / SUBSCRIBE_DONE / SUBSCRIBE_ERROR write can complete its writeMutex.withLock { ... } critical section. Previously cancellation could truncate a frame mid-flight and we'd send FIN over a corrupted stream. - #9 pumps wrapped in try/catch that calls close(...) on unexpected exceptions, so a transport-died-mid-session no longer leaves the session thinking it's healthy with new subscribe/announce calls hanging on a dead peer. - #10 TrackPublisher.send rolls back nextObjectId when every datagram fan-out fails (transport down). The audio-rooms NIP wants strictly contiguous object ids per group; a gap from a fully-failed send would trip strict subscribers. - #13 DefaultNestsSpeaker.close drops `gate` before calling activeHandle.close() / session.close(). The teardown chain runs cancelAndJoin on the broadcaster + sends SUBSCRIBE_DONE per attached subscriber + joins MoQ pumps; holding the gate through all of that blocked any other concurrent API call on this speaker. Resource lifecycle (audit MoQ #11/#12): - AudioRoomBroadcaster.stop() now `cancelAndJoin`s the loop before releasing the encoder + closing the publisher. The loop's last encoder.encode/publisher.send no longer races encoder.release()/publisher.close() — both produced use-after-release on native MediaCodec on Android, the latter sent orphan OBJECT_DATAGRAMs to subscribers we'd just told SUBSCRIBE_DONE. - AudioRoomPlayer.stop() promoted to `suspend` + cancelAndJoin for the same reason: decoder.release() ran while the decode loop was still inside MediaCodec.decode(...), undefined behaviour. Updated VM call sites (closeSubscription, teardown) to route both player.stop() and handle.unsubscribe() through one launched coroutine via the new `detach(): Pair<AudioRoomPlayer?, SubscribeHandle?>` shape. ViewModel fixes: - #4 auto-retry uses a single `retryPending: Boolean` flag instead of `Job.isActive`. Two scheduleAutoRetry calls could previously both pass `Job.isActive == false` (the launched body had just started) and stack a second retry on top of one already running. - #7 setMicMuted updates the UI INSIDE the launched coroutine, after the suspending broadcastHandle.setMuted() returns. Previously the indicator could claim "muted" while audio was still on the wire if the handle's setMuted suspended on a gate. - #8 connect() cancels the previous stateObserverJob before kicking off the new attempt, so a delayed Failed/Closed emission from the old listener can no longer clobber the fresh Connecting UI. - #9 disconnect() clears requestedSpeakers, so a fresh connect() to a different room (or the same room after a long pause) doesn't reuse a stale speaker snapshot. - #12 updateSpeakers filters out the user's own pubkey: subscribing to your own forwarded audio would echo through the local playback device whenever the broadcast track loops back from the relay. Android lifecycle / PIP / service: - #6 PIP aspect ratio flipped from 9:16 (portrait sliver) to 16:9 (landscape) so the row of avatars under the title actually fits. - #7 process-death recovery: when AudioRoomBridge is empty (previous process's AccountViewModel is gone), redirect to MainActivity before finish() so the user lands somewhere meaningful instead of a black-flash. - #8 AudioRoomForegroundService.onStartCommand always calls startForeground first, on every invocation including ACTION_STOP. startForegroundService's 5-second contract requires it; previously the STOP path skipped it. startForeground itself wrapped in runCatching so a foreground-not-allowed exception bails cleanly rather than leaking the wake-lock. - #10 wake-lock timeout reduced 12 h → 4 h. Stuck connections that fail to detect a network drop no longer hold the device awake for half a day. - #11 presence-event spam fix: split the publish loop into a heartbeat keyed only on (address, handRaised) and a separate debounced state-change publisher keyed on micMutedTag. Every mute toggle previously triggered a full sign + publish + relay round trip; now we coalesce within a 500 ms window. - #12 final "leaving" presence routed through GlobalScope.launch instead of rememberCoroutineScope (which is cancelled on dispose, so the leave event almost never reached the relay). - #14 RECORD_AUDIO denial recovery: when the user has tapped "Don't ask again", the launcher silently returns false. New "Open settings" button deep-links to the app's settings page so the user can re-grant the permission and try again. - #15 setPictureInPictureParams now updates outside PIP too, so the next entry shows the correct mute-state icon without an extra flip. - #16 onNewIntent override: a second Join tap for a different room finishes the current Activity and starts a fresh one with the new extras, instead of singleTask silently keeping the old room running. Verified: spotlessApply clean; :commons:jvmTest, :nestsClient:jvmTest (80 tests), :amethyst:compilePlayDebugKotlin all green. Audit findings still deferred (all documented inline / in this commit): - MoQ #1, #8: wire-format draft pinning (draft-17 vs draft-11). Needs the M4 manual interop pass to confirm what nests is actually speaking. - VM #6: confined map mutation under a single dispatcher. Current setup (Dispatchers.Main via setMain in tests, viewModelScope in prod) is functionally fine; full belt-and-suspenders confining is a separate concurrency review. - VM #10: test coverage gaps for retry + speaker reconcile cycle + setMicMuted no-handle case + server-initiated Closed leaves stale state. Each is a dedicated test. - Android #18: startListening / promoteToMicrophone race. Mitigation (always declaring microphone foreground type) requires unconditional RECORD_AUDIO grant which listener-only users won't have.
This commit is contained in:
Claude
@@ -199,13 +199,23 @@ class DefaultNestsSpeaker internal constructor(
|
||||
}
|
||||
|
||||
override suspend fun close() {
|
||||
// Take the active handle under `gate` (so a concurrent
|
||||
// `startBroadcasting` can't observe a half-closed state), then
|
||||
// release the lock before calling `handle.close()` and
|
||||
// `session.close()` — both are long-running suspend operations
|
||||
// (broadcaster.stop awaits cancelAndJoin; session.close fires
|
||||
// SUBSCRIBE_DONE per attached subscriber and joins pumps). Holding
|
||||
// the gate through them would block any other API call on this
|
||||
// speaker for the entire teardown duration.
|
||||
val handle: DefaultBroadcastHandle?
|
||||
gate.withLock {
|
||||
if (state.value is NestsSpeakerState.Closed) return
|
||||
activeHandle?.let { runCatching { it.close() } }
|
||||
handle = activeHandle
|
||||
activeHandle = null
|
||||
runCatching { session.close() }
|
||||
mutableState.value = NestsSpeakerState.Closed
|
||||
}
|
||||
handle?.runCatching { close() }
|
||||
runCatching { session.close() }
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+9
-1
@@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.moq.MoqSession
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.cancelAndJoin
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
/**
|
||||
@@ -129,11 +130,18 @@ class AudioRoomBroadcaster(
|
||||
* Stop the loop, release the mic, release the encoder, close the MoQ
|
||||
* publisher (which fires SUBSCRIBE_DONE to every attached subscriber).
|
||||
* Idempotent.
|
||||
*
|
||||
* Implementation note: we `cancelAndJoin` the loop before releasing
|
||||
* the encoder and closing the publisher. Otherwise the loop's last
|
||||
* `encoder.encode(...)` or `publisher.send(...)` could race
|
||||
* `encoder.release()` / `publisher.close()` and produce orphan
|
||||
* OBJECT_DATAGRAMs to subscribers that already received SUBSCRIBE_DONE,
|
||||
* or use-after-release on the native MediaCodec.
|
||||
*/
|
||||
suspend fun stop() {
|
||||
if (stopped) return
|
||||
stopped = true
|
||||
job?.cancel()
|
||||
job?.cancelAndJoin()
|
||||
runCatching { capture.stop() }
|
||||
runCatching { encoder.release() }
|
||||
runCatching { publisher.close() }
|
||||
|
||||
+12
-3
@@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.moq.MoqObject
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.cancelAndJoin
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
@@ -101,11 +102,19 @@ class AudioRoomPlayer(
|
||||
}
|
||||
}
|
||||
|
||||
/** Stop playback, cancel the decode loop, release the decoder. Idempotent. */
|
||||
fun stop() {
|
||||
/**
|
||||
* Stop playback, cancel the decode loop, release the decoder. Idempotent.
|
||||
*
|
||||
* Suspending so callers can await the loop's exit before releasing
|
||||
* native resources. Calling `decoder.release()` while another coroutine
|
||||
* is mid-`decoder.decode(...)` is undefined behaviour for MediaCodec
|
||||
* (and most native decoders); `cancelAndJoin` waits for the loop to
|
||||
* unwind through its CancellationException path before we proceed.
|
||||
*/
|
||||
suspend fun stop() {
|
||||
if (stopped) return
|
||||
stopped = true
|
||||
job?.cancel()
|
||||
job?.cancelAndJoin()
|
||||
runCatching { player.stop() }
|
||||
runCatching { decoder.release() }
|
||||
}
|
||||
|
||||
@@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.moq.MoqSession.Companion.client
|
||||
import com.vitorpamplona.nestsclient.moq.MoqSession.Companion.server
|
||||
import com.vitorpamplona.nestsclient.transport.WebTransportBidiStream
|
||||
import com.vitorpamplona.nestsclient.transport.WebTransportSession
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
@@ -356,13 +357,31 @@ class MoqSession private constructor(
|
||||
// ---------------------------------------------------------------- Pumps
|
||||
|
||||
private fun startPumps() {
|
||||
// If a pump exits unexpectedly (transport died, peer hung up), the
|
||||
// session is no longer healthy — failing in-flight subscribe/announce
|
||||
// waiters and flipping `closed=true` prevents new operations from
|
||||
// hanging on a peer that will never reply. The control pump is the
|
||||
// primary signal for transport death; the datagram pump exiting on
|
||||
// its own is rare but treated the same way.
|
||||
controlPumpJob =
|
||||
pumpScope.launch {
|
||||
runControlPump()
|
||||
try {
|
||||
runControlPump()
|
||||
} catch (ce: CancellationException) {
|
||||
throw ce
|
||||
} catch (t: Throwable) {
|
||||
runCatching { close(0, "control pump failed: ${t.message}") }
|
||||
}
|
||||
}
|
||||
datagramPumpJob =
|
||||
pumpScope.launch {
|
||||
runDatagramPump()
|
||||
try {
|
||||
runDatagramPump()
|
||||
} catch (ce: CancellationException) {
|
||||
throw ce
|
||||
} catch (t: Throwable) {
|
||||
runCatching { close(0, "datagram pump failed: ${t.message}") }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -424,16 +443,41 @@ class MoqSession private constructor(
|
||||
}
|
||||
|
||||
is AnnounceError -> {
|
||||
val deferred =
|
||||
// Two cases:
|
||||
// (a) ANNOUNCE_ERROR arrives BEFORE ANNOUNCE_OK — the peer
|
||||
// refused our announce. The deferred is still pending;
|
||||
// fail it and remove the announces[] entry we
|
||||
// optimistically inserted in announce().
|
||||
// (b) ANNOUNCE_ERROR arrives AFTER ANNOUNCE_OK — the peer
|
||||
// is revoking a previously-accepted announce
|
||||
// (session-level kick, e.g. policy change). Mark the
|
||||
// handle session-closed so further sends/openTracks
|
||||
// short-circuit, but don't remove from announces[]
|
||||
// since unannounce() is still the canonical removal
|
||||
// path and inbound SUBSCRIBE must continue to see the
|
||||
// namespace as "withdrawn" rather than missing.
|
||||
val handle =
|
||||
stateMutex.withLock {
|
||||
announces.remove(msg.namespace)
|
||||
pendingAnnounces.remove(msg.namespace)
|
||||
val pending = pendingAnnounces.remove(msg.namespace)
|
||||
if (pending != null) {
|
||||
announces.remove(msg.namespace)
|
||||
pending.completeExceptionally(
|
||||
MoqProtocolException(
|
||||
"ANNOUNCE rejected: code=${msg.errorCode} reason=${msg.reasonPhrase}",
|
||||
),
|
||||
)
|
||||
null
|
||||
} else {
|
||||
announces[msg.namespace]
|
||||
}
|
||||
}
|
||||
deferred?.completeExceptionally(
|
||||
MoqProtocolException(
|
||||
"ANNOUNCE rejected: code=${msg.errorCode} reason=${msg.reasonPhrase}",
|
||||
),
|
||||
)
|
||||
handle?.let { existing ->
|
||||
stateMutex.withLock { existing.markSessionClosedLocked() }
|
||||
runCatching {
|
||||
writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(msg.namespace))) }
|
||||
}
|
||||
stateMutex.withLock { announces.remove(msg.namespace) }
|
||||
}
|
||||
}
|
||||
|
||||
is Subscribe -> {
|
||||
@@ -573,8 +617,18 @@ class MoqSession private constructor(
|
||||
inboundSubscribers.clear()
|
||||
publisherSubscribers.clear()
|
||||
}
|
||||
controlPumpJob?.cancel()
|
||||
datagramPumpJob?.cancel()
|
||||
// Cancel pumps then JOIN them, so any in-flight pump write
|
||||
// (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.
|
||||
val ctrlPump = controlPumpJob
|
||||
val datagramPump = datagramPumpJob
|
||||
ctrlPump?.cancel()
|
||||
datagramPump?.cancel()
|
||||
runCatching { ctrlPump?.join() }
|
||||
runCatching { datagramPump?.join() }
|
||||
runCatching { writeMutex.withLock { controlStream.finish() } }
|
||||
runCatching { transport.close(code, reason) }
|
||||
}
|
||||
@@ -673,13 +727,24 @@ class MoqSession private constructor(
|
||||
sessionClosed = true
|
||||
toClose = tracks.values.toList()
|
||||
tracks.clear()
|
||||
announces.remove(namespace)
|
||||
// Note: keep `announces[namespace] = this` until UNANNOUNCE
|
||||
// is on the wire. Inbound SUBSCRIBE during this window will
|
||||
// see `sessionClosed=true` via publisherForLocked → null →
|
||||
// we reply SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST), which is
|
||||
// accurate (no track to serve). Removing now would race
|
||||
// with the control pump and leave a window where the peer
|
||||
// thinks the namespace exists but our state has dropped it.
|
||||
}
|
||||
toClose.forEach { runCatching { it.close() } }
|
||||
if (closed) return
|
||||
runCatching {
|
||||
writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(namespace))) }
|
||||
if (!closed) {
|
||||
runCatching {
|
||||
writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(namespace))) }
|
||||
}
|
||||
}
|
||||
// Now safe to forget the namespace entirely — UNANNOUNCE is on
|
||||
// the wire so any later inbound SUBSCRIBE was sent without
|
||||
// knowing the namespace was withdrawn.
|
||||
stateMutex.withLock { announces.remove(namespace) }
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -728,6 +793,16 @@ class MoqSession private constructor(
|
||||
runCatching { transport.sendDatagram(datagram) }
|
||||
.onSuccess { anySent = true }
|
||||
}
|
||||
// Roll the objectId back if the entire fan-out failed (e.g.
|
||||
// transport down). Audio-rooms NIP wants strictly contiguous
|
||||
// object ids per group; a gap from a fully-failed send would
|
||||
// trip strict subscribers. We roll back only if the next send
|
||||
// hasn't already grabbed an id past us.
|
||||
if (!anySent) {
|
||||
stateMutex.withLock {
|
||||
if (nextObjectId == objectId + 1) nextObjectId = objectId
|
||||
}
|
||||
}
|
||||
return anySent
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user