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

test(nests): pure unit tests for cliff-detector predicate + diagnostics + Connected restart

Browse Source 
The two-phone repro on commit cb61082 showed the cliff still triggers
at ~13 s with framesPerGroup=10 (last drainOneGroup at ts 14:26:14,
broadcaster keeps pushing through 14:26:25+ unimpeded), but the cliff
detector did NOT fire — no `cliff-detector: announced+subscribed but
silent…` log in the receiver's NestRx output despite the gap clearly
exceeding the 4 s threshold. Three changes here narrow the gap.

1. Extract `computeStalledSpeakers` as a pure top-level internal
   function over `Map<String, TimeMark>`. The detector loop now calls
   it; the rest of the predicate logic (cooldown, announced ∩ active
   ∩ stale, ≥ inclusive boundary) is now testable without standing up
   the NestViewModel coroutine machinery.

2. New `CliffDetectorTest` (commonTest, 12 tests) drives
   `computeStalledSpeakers` with a `kotlin.time.TestTimeSource`.
   Pins the boundary cases the next refactor would otherwise drift
   on: at-threshold inclusive, just-under empty, no-frame-yet ignored,
   not-announced ignored, multi-speaker classification, cooldown
   suppress, cooldown release, and a few more. All 12 pass.

3. Add diagnostic logging to the live cliff detector. The two-phone
   logs proved the predicate's logic works (the `lastFrameAt` mark
   was clearly aged past the threshold), so the silence has to be
   in the gating: `listener` null, connection state not Connected,
   or `_announcedSpeakers` not containing the pubkey at scan time.
   New logs:
     - "cliff-detector launched" once on start
     - "cliff-detector tick=N gated: listener=… conn=…" every 5 s
       when gated
     - "cliff-detector tick=N active=… announced=… lastFrameAges=…"
       every 5 s when running
     - "cliff-detector EXITED after N ticks (closed=…)" on cancel
   so the next repro tells us which gate is failing.

4. Defensive restart on `NestsListenerState.Connected`. The Closed
   branch in `observeListenerState` calls `teardown(targetState =
   Closed)`, which cancels `cliffDetectorJob` along with the rest of
   the per-session jobs. If the cliff detector itself is what
   triggered the recycle, the wrapper emits Closed → Reconnecting
   → Connected; we'd come back online with a dead detector and the
   next cliff event would slip past undetected. Calling
   `startCliffDetector()` (idempotent) on every Connected entry
   keeps the detector alive across the cycle.

Defaults stay where they were: 4 s `ROOM_AUDIO_CLIFF_TIMEOUT_MS`,
1 s scan, 8 s cooldown — the unit tests also pin those default
constants by relying on the function's defaults in two of the cases.
This commit is contained in:
Claude
2026-05-05 18:54:25 +00:00
parent cb61082bc4
commit f2205dc97e
2 changed files with 438 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt
+129 -35
View File
@@ -277,7 +277,7 @@ class NestViewModel(
* subscribe bidi is still open from QUIC's POV — meaning no other
* layer notices.
*/
private val lastFrameAt = mutableMapOf<String, kotlin.time.TimeSource.Monotonic.ValueTimeMark>()
private val lastFrameAt = mutableMapOf<String, kotlin.time.TimeMark>()
/**
* 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<String>,
announcedSpeakers: Set<String>,
lastFrameAt: Map<String, kotlin.time.TimeMark>,
lastRecycleAt: kotlin.time.TimeMark?,
cliffTimeoutMs: Long = ROOM_AUDIO_CLIFF_TIMEOUT_MS,
cooldownMs: Long = ROOM_AUDIO_CLIFF_COOLDOWN_MS,
): List<String> {
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
commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/CliffDetectorTest.kt
+309
View File
@@ -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)
}
}