diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt index bff80eb3e..efd73e927 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt @@ -277,7 +277,7 @@ class NestViewModel( * subscribe bidi is still open from QUIC's POV — meaning no other * layer notices. */ - private val lastFrameAt = mutableMapOf() + private val lastFrameAt = mutableMapOf() /** * Periodic scan of [activeSubscriptions] vs [_announcedSpeakers] @@ -298,7 +298,7 @@ class NestViewModel( * the new session — the new session has no incoming frames yet, * which would otherwise trip the detector again immediately. */ - private var lastCliffRecycleAt: kotlin.time.TimeSource.Monotonic.ValueTimeMark? = null + private var lastCliffRecycleAt: kotlin.time.TimeMark? = null /** * Per-speaker catalog-fetch coroutines. Each entry is the @@ -851,6 +851,18 @@ class NestViewModel( // this is typically a no-op in toAdd / toRemove // — runs anyway for the first-Connected path. reconcileSubscriptions() + // Defensive restart: if the wrapper went + // through a Closed transient (e.g. the cliff + // detector itself triggered `recycleSession()`, + // which closes the inner listener and waits + // for the orchestrator to reopen), our + // [Closed] branch below cancelled the + // detector via [teardown]. We need it back + // up on the fresh session — otherwise a + // second cliff event during the same room + // sits undetected. Idempotent: returns + // immediately if a job is already active. + startCliffDetector() } is NestsListenerState.Failed -> { @@ -1295,42 +1307,68 @@ class NestViewModel( */ private fun startCliffDetector() { if (cliffDetectorJob?.isActive == true) return + com.vitorpamplona.quartz.utils.Log + .d("NestRx") { "cliff-detector launched" } cliffDetectorJob = viewModelScope.launch { - while (true) { - delay(ROOM_AUDIO_CLIFF_CHECK_INTERVAL_MS) - if (closed) return@launch - val l = listener ?: continue - if (_uiState.value.connection !is ConnectionUiState.Connected) continue - // Cooldown: if we very recently triggered a recycle, - // give the orchestrator time to bring up the new - // session before re-checking. A new session has - // nothing in `lastFrameAt` yet which would otherwise - // immediately re-trigger. - val sinceRecycle = lastCliffRecycleAt?.elapsedNow() - if (sinceRecycle != null && - sinceRecycle.inWholeMilliseconds < ROOM_AUDIO_CLIFF_COOLDOWN_MS - ) { - continue + var ticks = 0L + try { + while (true) { + delay(ROOM_AUDIO_CLIFF_CHECK_INTERVAL_MS) + ticks += 1 + if (closed) return@launch + val l = listener + val connState = _uiState.value.connection + if (l == null || connState !is ConnectionUiState.Connected) { + // Periodic visibility into "why isn't the cliff + // detector acting" — the receiver-side cliff + // we're chasing has been silent in production + // logs even when the wire conditions look right, + // so dump our gating state every 5 s. + if (ticks % CLIFF_DIAG_LOG_EVERY == 0L) { + com.vitorpamplona.quartz.utils.Log.d("NestRx") { + "cliff-detector tick=$ticks gated: listener=${l != null} conn=${connState::class.simpleName}" + } + } + continue + } + val activeSpeakers = + activeSubscriptions + .asSequence() + .filter { (_, slot) -> slot.isPlaying } + .map { it.key } + .toSet() + val announced = _announcedSpeakers.value + if (ticks % CLIFF_DIAG_LOG_EVERY == 0L) { + val ages = + lastFrameAt.entries.joinToString { (k, mark) -> + "${k.take(8)}=${mark.elapsedNow().inWholeMilliseconds}ms" + } + com.vitorpamplona.quartz.utils.Log.d("NestRx") { + "cliff-detector tick=$ticks active=${activeSpeakers.size} announced=${announced.size} lastFrameAges=[$ages]" + } + } + val stalled = + computeStalledSpeakers( + activeSpeakers = activeSpeakers, + announcedSpeakers = announced, + lastFrameAt = lastFrameAt, + lastRecycleAt = lastCliffRecycleAt, + cliffTimeoutMs = ROOM_AUDIO_CLIFF_TIMEOUT_MS, + cooldownMs = ROOM_AUDIO_CLIFF_COOLDOWN_MS, + ) + if (stalled.isEmpty()) continue + com.vitorpamplona.quartz.utils.Log.w("NestRx") { + "cliff-detector: announced+subscribed but silent for ≥${ROOM_AUDIO_CLIFF_TIMEOUT_MS}ms — recycling session. stalled=$stalled" + } + lastCliffRecycleAt = + kotlin.time.TimeSource.Monotonic + .markNow() + runCatching { l.recycleSession() } } - val announced = _announcedSpeakers.value - val stalled = - activeSubscriptions - .asSequence() - .filter { (pubkey, slot) -> slot.isPlaying && pubkey in announced } - .filter { (pubkey, _) -> - val mark = lastFrameAt[pubkey] ?: return@filter false - mark.elapsedNow().inWholeMilliseconds >= ROOM_AUDIO_CLIFF_TIMEOUT_MS - }.map { it.key } - .toList() - if (stalled.isEmpty()) continue - com.vitorpamplona.quartz.utils.Log.w("NestRx") { - "cliff-detector: announced+subscribed but silent for ≥${ROOM_AUDIO_CLIFF_TIMEOUT_MS}ms — recycling session. stalled=$stalled" - } - lastCliffRecycleAt = - kotlin.time.TimeSource.Monotonic - .markNow() - runCatching { l.recycleSession() } + } finally { + com.vitorpamplona.quartz.utils.Log + .w("NestRx") { "cliff-detector EXITED after $ticks ticks (closed=$closed)" } } } } @@ -1634,6 +1672,62 @@ const val ROOM_AUDIO_CLIFF_CHECK_INTERVAL_MS: Long = 1_000L */ const val ROOM_AUDIO_CLIFF_COOLDOWN_MS: Long = 8_000L +/** + * Diagnostic log frequency for the cliff detector. Emit a state-dump + * line every Nth 1-second tick — i.e. every 5 s while connected and + * idle, so logcat captures the "active / announced / lastFrameAges" + * snapshot we use to debug silent-cliff cases without flooding when + * everything is fine. + */ +private const val CLIFF_DIAG_LOG_EVERY: Long = 5L + +/** + * Pure logic of the cliff detector — extracted so headless tests can + * exercise it with a [kotlin.time.TestTimeSource] without standing up + * the full [NestViewModel] coroutine machinery. + * + * Returns the list of pubkeys that should trigger a `recycleSession()` + * RIGHT NOW. Empty list means "no cliff observed; do nothing on this + * tick." + * + * Recycle conditions, all-of: + * - the pubkey has an active subscription on our side + * ([activeSpeakers]) — i.e. we're actively pulling its audio, + * - the relay has told us this pubkey is broadcasting + * ([announcedSpeakers]), + * - we've received at least one MoQ object in the past + * ([lastFrameAt] has an entry — we don't preemptively recycle + * a brand-new subscription that simply hasn't ramped up yet), + * - the elapsed time since that last object is at or past + * [cliffTimeoutMs]. + * + * Suppression: + * - if [lastRecycleAt] is non-null and less than [cooldownMs] has + * elapsed since it, returns empty (no cascading recycles while + * the wrapper is mid-handshake on a fresh session). + */ +internal fun computeStalledSpeakers( + activeSpeakers: Set, + announcedSpeakers: Set, + lastFrameAt: Map, + lastRecycleAt: kotlin.time.TimeMark?, + cliffTimeoutMs: Long = ROOM_AUDIO_CLIFF_TIMEOUT_MS, + cooldownMs: Long = ROOM_AUDIO_CLIFF_COOLDOWN_MS, +): List { + if (lastRecycleAt != null && + lastRecycleAt.elapsedNow().inWholeMilliseconds < cooldownMs + ) { + return emptyList() + } + return activeSpeakers + .asSequence() + .filter { it in announcedSpeakers } + .filter { pubkey -> + val mark = lastFrameAt[pubkey] ?: return@filter false + mark.elapsedNow().inWholeMilliseconds >= cliffTimeoutMs + }.toList() +} + /** * How long a kind-7 reaction stays in * [NestViewModel.recentReactions] before the eviction sweep diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/CliffDetectorTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/CliffDetectorTest.kt new file mode 100644 index 000000000..4b8f285c6 --- /dev/null +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/CliffDetectorTest.kt @@ -0,0 +1,309 @@ +/* + * 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 kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.ExperimentalTime +import kotlin.time.TestTimeSource + +/** + * Unit tests for the [computeStalledSpeakers] predicate that drives the + * listener-side cliff detector in [NestViewModel]. The predicate is + * extracted as a pure function specifically so we can drive it with a + * [TestTimeSource] — neither `runTest`'s virtual scheduler nor a real + * monotonic clock would let us deterministically place events relative + * to the 4 s `ROOM_AUDIO_CLIFF_TIMEOUT_MS` threshold. + * + * The detector defends against the moq-rs production-relay forward- + * queue cliff documented in + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`: + * the relay opens fewer (or zero) new uni streams to a slow + * subscriber while the announce stream still says the broadcast is + * Active and the subscribe bidi is still alive at QUIC level — no + * other layer notices. These tests pin the recycle gate so a future + * refactor doesn't accidentally widen or narrow when we recycle. + */ +@OptIn(ExperimentalTime::class) +class CliffDetectorTest { + @Test + fun returnsEmptyWhenNoSpeakersActive() { + val ts = TestTimeSource() + val result = + computeStalledSpeakers( + activeSpeakers = emptySet(), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to ts.markNow()), + lastRecycleAt = null, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun ignoresSpeakerNotInAnnounced() { + // A speaker we're subscribed to but who isn't in the relay's + // announce-stream output is most likely either (a) already + // off stage and the relay just hasn't propagated the Ended + // yet, or (b) in the brief gap between an Ended and the + // wrapper re-issuing. Recycling the whole transport here + // would be wasteful — we only act on confirmed cliffs. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 10_000.milliseconds // way past threshold + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = emptySet(), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = null, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun ignoresSpeakerWithNoFrameYet() { + // Brand-new subscription that hasn't ramped up: lastFrameAt + // has no entry. Don't recycle — the wrapper's per-speaker + // re-issue + opener-throws backoff already retries on its + // own; the cliff detector only acts once we've proven the + // subscription was working. + val ts = TestTimeSource() + ts += 10_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = emptyMap(), + lastRecycleAt = null, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun returnsEmptyJustBeforeThreshold() { + // Boundary case: 1 ms under the 4 s threshold. The detector + // should still consider this healthy. Relevant because audio + // groups arrive at ~100 ms cadence (framesPerGroup=10), so a + // false positive at ≈ threshold would recycle on every + // group-rollover hiccup. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 3_999.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = null, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun returnsStalledSpeakerAtThresholdInclusive() { + // Exactly at threshold: include. The check is `>=`, not `>`. + // Tested explicitly because boundary semantics here are + // user-visible (this is the "audio went silent" detection). + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = null, + ) + assertEquals(listOf(ALICE), result) + } + + @Test + fun returnsStalledSpeakerWellPastThreshold() { + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 30_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = null, + ) + assertEquals(listOf(ALICE), result) + } + + @Test + fun returnsAllStalledSpeakersAtOnceMixedWithFreshOnes() { + // Multi-speaker stage: alice and bob have stalled, charlie's + // subscription is fresh (just received a frame). Recycle + // should fire for the stalled set and leave charlie alone — + // the `recycleSession` itself takes the whole listener down, + // but the test here is the predicate's classification. + val ts = TestTimeSource() + val aliceFrame = ts.markNow() + val bobFrame = ts.markNow() + ts += 3_000.milliseconds + val charlieFrame = ts.markNow() + ts += 3_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE, BOB, CHARLIE), + announcedSpeakers = setOf(ALICE, BOB, CHARLIE), + lastFrameAt = + mapOf( + ALICE to aliceFrame, + BOB to bobFrame, + CHARLIE to charlieFrame, + ), + lastRecycleAt = null, + ) + assertEquals(setOf(ALICE, BOB), result.toSet()) + } + + @Test + fun cooldownSuppressesRecycleEvenWhenStalled() { + // After a recycle, the wrapper opens a fresh QUIC transport. + // The new session has no `lastFrameAt` entries yet for any + // pubkey; without a cooldown we would re-trigger on the + // very next 1 s tick because the prior tick's stalled + // pubkeys still age past the threshold (their lastFrameAt + // hasn't been updated by the new session yet). 8 s cooldown + // covers the typical reconnect handshake. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_500.milliseconds // past threshold + val recycleMark = ts.markNow() // recycle just fired + ts += 1_000.milliseconds // 1 s into cooldown + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun cooldownReleasesAfterTimeoutPasses() { + // Once cooldown elapses, a still-stalled subscription + // becomes eligible to recycle again. Important for the + // case where the recycle didn't actually fix the cliff — + // we want a second attempt rather than getting wedged. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_500.milliseconds + val recycleMark = ts.markNow() + ts += 8_001.milliseconds // 1 ms past cooldown + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + ) + assertEquals(listOf(ALICE), result) + } + + @Test + fun customTimeoutsAreHonored() { + // Defaults are wired into the production VM, but the function + // accepts overrides so a test for tighter / looser tolerances + // can drive it without mutating the constants. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 1_000.milliseconds + + val withDefault = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = null, + ) + assertTrue(withDefault.isEmpty(), "1 s elapsed shouldn't trip 4 s default") + + val withTightTimeout = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = null, + cliffTimeoutMs = 500L, + ) + assertEquals(listOf(ALICE), withTightTimeout, "1 s elapsed should trip a 500 ms threshold") + } + + @Test + fun activeSpeakerNotInLastFrameAtIsIgnoredEvenWithStalledPeers() { + // Mixed: alice is announced + active + stalled, bob is + // announced + active but has no frame yet. Detector should + // return alice only — bob hasn't proven the subscription + // works, so it's not "stalled" yet, just slow to start. + val ts = TestTimeSource() + val aliceFrame = ts.markNow() + ts += 5_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE, BOB), + announcedSpeakers = setOf(ALICE, BOB), + lastFrameAt = mapOf(ALICE to aliceFrame), + lastRecycleAt = null, + ) + assertEquals(listOf(ALICE), result) + } + + @Test + fun nullLastRecycleNeverSuppresses() { + // Null `lastRecycleAt` means "we've never recycled" — the + // very first cliff event of a session must fire. The + // cooldown check is gated on non-null. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 5_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = null, + ) + assertEquals(listOf(ALICE), result) + } + + companion object { + private val ALICE = "a".repeat(64) + private val BOB = "b".repeat(64) + private val CHARLIE = "c".repeat(64) + } +}