refactor(nests): extract ActiveSubscription + plan deferred manager refactor (Audit-9, Audit-14)

Audit-9: NestViewModel.kt is 2112 lines and growing, ~1000 of which are subscription-lifecycle state machine concerns intertwined with the room-level public API. Pulling out the full `NestSubscriptionManager` is a multi-week refactor with subtle coupling (catalog readiness affects spinner state, mute has effective + per-speaker flavours, expiry jobs need the parent scope) and warrants its own focused review pass. For now, take the small tractable subset: - Extract `ActiveSubscription` from a `private inner class` in NestViewModel to its own file as `internal class ActiveSubscription` in the same package. The class is purely state-holding (handle, roomPlayer, player, isPlaying); zero VM coupling beyond the slot map's value type. Same visibility for NestViewModel callers; one less private helper class buried 1500 lines into NestViewModel.kt. - File `commons/plans/2026-05-06-nest-subscription-manager-extraction.md` documents the deferred full extraction: target shape, state that moves, methods that move, what stays in VM, why deferred, when to land. Picks up the next person who opens NestViewModel.kt rather than leaving them to re-derive the rationale. Audit-14: T11.3 (stream priority for moq-lite group uni streams) was deferred from the T11 commit (drop bestEffort=true) because the :quic-side change touches the writer's hot path and warrants its own review pass. New file `nestsClient/plans/2026-05-06-stream-priority-followup.md` spells out: - Why: bestEffort=true was incidentally biasing drain order toward newer groups; without it, the writer's round-robin order can serve a stale group when a fresh one is more useful. - Target shape: `QuicStream.priority` field, sortedByDescending in the writer's send-frame loop, `WebTransportWriteStream.setPriority` pass-through, `MoqLiteSession.openGroupStream` calls setPriority(sequence). With code sketches. - Test: pin iteration order via the writer's emitted-frames tape. - Risk profile: starvation, perf cost of per-pass sort, compat. - When to land: after interop verification stabilises. No code changes in this commit beyond the ActiveSubscription move. https://claude.ai/code/session_014JfZJHSTvyYYWJbC9VbB47
2026-05-06 20:04:56 +00:00
parent e9d19e5de4
commit 28358b4141
4 changed files with 353 additions and 43 deletions
@@ -0,0 +1,109 @@
+# Extract `NestSubscriptionManager` from `NestViewModel`
+
+**Status**: deferred — flagged in the audit pass, not landed yet.
+
+## Why
+
+`NestViewModel.kt` is 2112 lines and growing. ~1000 of those lines are
+the per-speaker subscription-lifecycle state machine: open / close /
+reconcile / catalog-fetch / decoder-await / level-tap / mute-routing /
+hush. The rest is room-level public API (connect, disconnect,
+broadcast, presence, reactions, focus / network observers, UI state).
+
+These two concerns are coupled by shared mutable state but do not
+share a *responsibility*. The audit-9 finding flagged it as the
+single biggest SRP breach in the touched code.
+
+We extracted `ActiveSubscription` into its own file (`ActiveSubscription.kt`)
+in commit `<TBD>` as a stepping stone — the deferred extraction
+proper is the orchestration class.
+
+## Target shape
+
+```kotlin
+internal class NestSubscriptionManager(
+    private val viewModelScope: CoroutineScope,
+    private val signer: NostrSigner,
+    private val decoderFactory: (channelCount: Int, sampleRate: Int) -> OpusDecoder,
+    private val playerFactory: (channelCount: Int, sampleRate: Int) -> AudioPlayer,
+    private val onActiveSpeakersChanged: (Set<String>) -> Unit,
+    private val onSpeakerActivity: (String) -> Unit,
+    private val onAudioLevel: (String, Float) -> Unit,
+    private val onConnectingSpeakerChanged: (pubkey: String, connecting: Boolean) -> Unit,
+    private val effectiveListenMuted: () -> Boolean,
+    private val locallyHushed: () -> Set<String>,
+) {
+    val speakerCatalogs: StateFlow<Map<String, RoomSpeakerCatalog>>
+    val audioLevels: StateFlow<Map<String, Float>>
+
+    fun bind(listener: NestsListener)
+    fun unbind()  // cancel everything; release native resources
+    fun updateSpeakers(requested: Set<String>)
+    fun applyEffectiveMute()
+    fun setLocalHushed(pubkey: String, hushed: Boolean)
+}
+```
+
+## State that moves
+
+From `NestViewModel`:
+- `activeSubscriptions: MutableMap<String, ActiveSubscription>`
+- `catalogJobs: MutableMap<String, Job>`
+- `_speakerCatalogs: MutableStateFlow<Map<String, RoomSpeakerCatalog>>`
+- `requestedSpeakers: Set<String>`
+- `speakingExpiryJobs: MutableMap<String, Job>` (for `onSpeakerActivity` debounce)
+- `_audioLevels` if separable
+
+## Methods that move
+
+- `reconcileSubscriptions`
+- `openSubscription` (~150 lines)
+- `closeSubscription`
+- `fetchSpeakerCatalog`
+- `awaitAudioPipelineConfig`
+- `onSpeakerActivity` debounce timer
+- `onAudioLevel` coalescing
+- `applyEffectiveListenMute`'s subscription-touching half
+- `setLocalHushed`'s player.setVolume half
+- `publishActiveSpeakers`
+
+## What stays in `NestViewModel`
+
+- Public lifecycle (`connect`, `disconnect`, `onCleared`)
+- Connection state machine (`ConnectionUiState`)
+- Broadcast state (`broadcast` / `_uiState.broadcast`)
+- Presence aggregation (kind 10312 events)
+- Reactions
+- Focus / network observers
+- The `NestUiState` composition itself
+
+`NestViewModel` calls `manager.bind(listener)` on each fresh
+listener and `manager.unbind()` on disconnect / cleanup.
+`updateSpeakers` and mute / hush propagate through to the manager.
+
+## Why deferred
+
+- 1000-line code move across ~10 methods + 5 state fields
+- Subtle coupling: catalog readiness affects spinner state, mute
+  state has effective + per-speaker flavours, expiry jobs need
+  parent scope, `closed` flag is currently a single VM-wide flag
+- Test surface: `NestViewModelTest` exercises subscription paths
+  through the VM today; would need to either keep that surface or
+  add a `NestSubscriptionManagerTest`.
+
+The extraction has a clean contract (callbacks for the bits that
+remain VM-side) but landing it without behavioural drift wants a
+focused review pass plus its own dedicated test rebuild — bigger
+than fits in the current audit-pass.
+
+## When to land
+
+After:
+- The catalog-driven decoder reconfig (T3) has run in production
+  for long enough that the `awaitAudioPipelineConfig` pattern is
+  proven against real publishers.
+- Any pending subscription-lifecycle bug fixes (e.g. the audit's
+  earlier "boundary-rebuild dangling decoder" path) are settled —
+  moving them mid-fix risks introducing regressions.
+
+Track as a follow-up issue rather than a near-term must-do.
@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.viewmodels
+
+import com.vitorpamplona.nestsclient.audio.AudioPlayer
+import com.vitorpamplona.nestsclient.audio.NestPlayer
+import com.vitorpamplona.nestsclient.moq.SubscribeHandle
+
+/**
+ * Per-pubkey state slot held in [NestViewModel.activeSubscriptions].
+ *
+ * Three lifecycle phases:
+ *   - **Pending**: just-reserved by `reconcileSubscriptions` to dedupe
+ *     concurrent reconciles. No handle / player attached yet. Constructor
+ *     via [pending].
+ *   - **Active**: [attach] wires the moq subscribe handle, the
+ *     [NestPlayer] decode loop, and the device player. [isPlaying]
+ *     flips true.
+ *   - **Detached**: [detach] returns the handle + roomPlayer pair so the
+ *     caller can run the suspending teardown (`NestPlayer.stop()` +
+ *     `SubscribeHandle.unsubscribe()`) in its own coroutine scope —
+ *     keeps the native MediaCodec / AudioTrack release ordered after
+ *     the decode loop has unwound (audit MoQ #11/#12).
+ *
+ * `internal` because only [NestViewModel]'s subscription-lifecycle
+ * paths (open / close / reconcile / mute / hush) construct or mutate
+ * these. Lifted to a top-level type rather than a nested class so the
+ * file split tracks concerns: this class is "subscription state
+ * machine"; the rest of NestViewModel is "ViewModel public surface +
+ * orchestration."
+ */
+internal class ActiveSubscription private constructor(
+    val pubkey: String,
+) {
+    private var handle: SubscribeHandle? = null
+    private var roomPlayer: NestPlayer? = null
+    var player: AudioPlayer? = null
+        private set
+    var isPlaying: Boolean = false
+        private set
+
+    fun attach(
+        handle: SubscribeHandle,
+        roomPlayer: NestPlayer,
+        player: AudioPlayer,
+    ) {
+        this.handle = handle
+        this.roomPlayer = roomPlayer
+        this.player = player
+        this.isPlaying = true
+    }
+
+    /**
+     * Hand the player + handle back to the caller's coroutine scope —
+     * `NestPlayer.stop()` and `SubscribeHandle.unsubscribe()` are
+     * both suspend, and the caller has the right scope to await them
+     * (so native MediaCodec/AudioTrack release runs after the decode
+     * loop has unwound, per audit MoQ #11/#12).
+     */
+    fun detach(): Pair<NestPlayer?, SubscribeHandle?> {
+        isPlaying = false
+        val p = roomPlayer
+        val h = handle
+        roomPlayer = null
+        handle = null
+        player = null
+        return p to h
+    }
+
+    companion object {
+        fun pending(pubkey: String) = ActiveSubscription(pubkey)
+    }
+}
@@ -42,7 +42,6 @@ import com.vitorpamplona.nestsclient.audio.OpusDecoder
 import com.vitorpamplona.nestsclient.audio.OpusEncoder
 import com.vitorpamplona.nestsclient.connectReconnectingNestsListener
 import com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker
-import com.vitorpamplona.nestsclient.moq.SubscribeHandle
 import com.vitorpamplona.nestsclient.transport.WebTransportFactory
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
 import com.vitorpamplona.quartz.nip53LiveActivities.chat.LiveActivitiesChatMessageEvent
@@ -1672,48 +1671,8 @@ class NestViewModel(
            }
    }

-    private class ActiveSubscription private constructor(
-        val pubkey: String,
-    ) {
-        private var handle: SubscribeHandle? = null
-        private var roomPlayer: NestPlayer? = null
-        var player: AudioPlayer? = null
-            private set
-        var isPlaying: Boolean = false
-            private set
-
-        fun attach(
-            handle: SubscribeHandle,
-            roomPlayer: NestPlayer,
-            player: AudioPlayer,
-        ) {
-            this.handle = handle
-            this.roomPlayer = roomPlayer
-            this.player = player
-            this.isPlaying = true
-        }
-
-        /**
-         * Hand the player + handle back to the caller's coroutine scope —
-         * `NestPlayer.stop()` and `SubscribeHandle.unsubscribe()` are
-         * both suspend, and the caller has the right scope to await them
-         * (so native MediaCodec/AudioTrack release runs after the decode
-         * loop has unwound, per audit MoQ #11/#12).
-         */
-        fun detach(): Pair<NestPlayer?, SubscribeHandle?> {
-            isPlaying = false
-            val p = roomPlayer
-            val h = handle
-            roomPlayer = null
-            handle = null
-            player = null
-            return p to h
-        }
-
-        companion object {
-            fun pending(pubkey: String) = ActiveSubscription(pubkey)
-        }
-    }
+    // ActiveSubscription was extracted to its own file in the same
+    // package — see [ActiveSubscription.kt].

    // Platform-specific Factory lives in `amethyst/.../nests/room/`,
    // not commonMain — the lifecycle KMP `ViewModelProvider.Factory`
@@ -0,0 +1,151 @@
+# Stream priority for moq-lite group uni streams (T11.3 follow-up)
+
+**Status**: deferred — flagged in T11 (drop `bestEffort=true`), not landed.
+
+## Why
+
+`T11` removed `bestEffort=true` from `MoqLiteSession.openGroupStream`
+because the QUIC contract it relied on (drop lost ranges silently
+without `RESET_STREAM`) created undeliverable streams that wedge
+peer reassembly buffers — the actual user-visible bug was a 30 s
+silent dropout per lost packet on lossy networks.
+
+`bestEffort=true` was incidentally providing a "newer-groups-skip-
+queued-retransmits" effect: the writer never queued retransmits in
+the first place, so under congestion the loss budget naturally
+biased toward dropping older lost ranges. With `bestEffort=true`
+gone, all retransmits are now queued, and under sustained
+congestion the writer drains streams in `streamRoundRobinStart`
+order — which can mean the listener catches up on a stale group
+when a fresh one would be more useful.
+
+The kixelated reference (`moq-rs`'s `Publisher::serve_group` in
+`rs/moq-lite/src/lite/publisher.rs:347-406`) addresses this by
+calling `stream.set_priority(priority.current())` on every group
+stream and biasing the writer toward higher-priority (newer)
+streams. We should do the same.
+
+## Target shape
+
+### `:quic` API
+
+Add a stable priority hook to `QuicStream`:
+
+```kotlin
+class QuicStream(...) {
+    @Volatile
+    var priority: Int = 0
+        // Higher drains first under contention. Default 0 = unchanged
+        // round-robin behaviour.
+}
+```
+
+Expose at the WebTransport layer:
+
+```kotlin
+interface WebTransportWriteStream {
+    fun setPriority(priority: Int)
+}
+```
+
+### `QuicConnectionWriter` — the load-bearing change
+
+Today's send-frame loop (`QuicConnectionWriter.kt:411-416`):
+
+```kotlin
+val streamsView = conn.streamsListLocked()
+val start = conn.streamRoundRobinStart % streamsView.size
+for (i in streamsView.indices) {
+    val stream = streamsView[(start + i) % streamsView.size]
+    // ...
+}
+```
+
+Replace with priority-then-round-robin:
+
+```kotlin
+val streamsView = conn.streamsListLocked()
+val sorted = streamsView.sortedByDescending { it.priority }
+val start = conn.streamRoundRobinStart % sorted.size
+for (i in sorted.indices) {
+    val stream = sorted[(start + i) % sorted.size]
+    // ...
+}
+```
+
+Sort is stable; same-priority streams retain insertion order, so the
+existing round-robin behaviour holds within a priority tier. Higher-
+priority streams always come first in the iteration.
+
+Cost: O(N log N) per drain pass, where N is active local-initiated
+streams. N is small (1–10 in the moq-lite audio path); the
+allocation of `sorted` per pass is the only real cost. If that
+shows up in profiling, switch to `kotlin.collections.IntArray`-
+backed indirect sort or maintain a priority-sorted list incrementally
+on `setPriority` calls.
+
+### moq-lite wiring
+
+`MoqLiteSession.openGroupStream:1022-1037`:
+
+```kotlin
+internal suspend fun openGroupStream(
+    subscribeId: Long,
+    sequence: Long,
+): WebTransportWriteStream {
+    val uni = transport.openUniStream()
+    uni.setPriority(sequence.coerceAtMost(Int.MAX_VALUE.toLong()).toInt())
+    uni.write(Varint.encode(MoqLiteDataType.Group.code))
+    uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence)))
+    return uni
+}
+```
+
+Newer groups have higher sequence → higher priority → drain first
+under congestion. The saturation conversion handles broadcasts that
+run long enough for `sequence` to exceed `Int.MAX_VALUE` (≈ 71
+years at our 1 group/sec production cadence; defensive only).
+
+## Test
+
+Add a unit test in `:quic` that builds a `QuicConnection`, opens
+two streams, sets `.priority = 0` on the first and `.priority = 1`
+on the second, queues bytes on both, and verifies the higher-
+priority stream's bytes hit the wire first. Doesn't need to fake
+real flow-control backpressure — pinning the iteration order via
+the writer's emitted-frames tape is sufficient to catch the
+regression case ("a later refactor accidentally re-introduces
+round-robin order").
+
+A second test in `:nestsClient` that opens 3 group streams and
+asserts later-sequence streams drain before earlier ones is
+nice-to-have but secondary; the `:quic`-level test pins the load-
+bearing invariant.
+
+## Why deferred
+
+The change is small in lines but touches the QUIC writer's hot
+path. Rebasing a future `:quic` retransmit / pacing change onto a
+priority-sorted iteration order is doable but the diff conflicts
+get noisy. Better landed as a focused PR after the current
+interop-work series stabilises.
+
+Risk profile:
+- **Bugs**: subtle starvation risk if a high-priority stream
+  always has `streamRemaining > 0` — round-robin tiebreaker within
+  a priority tier mitigates this for same-priority case, but the
+  cross-tier case needs deliberate thought (do we want strict
+  priority or weighted?).
+- **Performance**: per-pass sort allocation. Negligible for N≤10
+  but worth measuring if N grows.
+- **Compat**: streams without an explicit priority default to 0,
+  matching today's behaviour. Existing tests should pass unchanged.
+
+## When to land
+
+After the catalog interop series is fully verified (production
+audio with no degradation under realistic conditions), and ideally
+alongside a `:quic` perf review pass that touches the same code
+path. Not blocking on any audio bug today — `T11`'s reliable-
+delivery fix is the actual correctness change; this is the
+spec-aligned hardening.