From 6e4df4aad4f9686b8f4524dfbba3a36afc761912 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 5 May 2026 22:13:12 +0000 Subject: [PATCH] =?UTF-8?q?fix(nests):=20cliff-detector=20=E2=80=94=20rese?= =?UTF-8?q?t=20lastFrameAt=20on=20recycle=20+=20bump=20cooldown=20to=2030s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production logs at commit ea08c43 (run 18:05:14) showed the cliff detector now correctly fires (`active=1 announced=1` after the channelFlow chain fix) — but it then thrashes the relay: 18:05:25.488 cliff #1 → recycle, session 1 had 4 groups (3 s of audio) 18:05:40.512 cliff #2 → recycle, session 2 had 36 groups (7 s of audio) 18:05:48.538 cliff #3 → recycle, session 3 had 0 groups, then ... 18:05:56.566 cliff #4 → recycle, session 4 can't even subscribe ("subscribe stream FIN before reply" — relay refusing) Two compounding issues: 1. `lastFrameAt[pubkey]` was NOT reset when the cliff detector triggered a recycle. The next 1-second tick after the 8 s cooldown re-evaluated `lastFrameAt[pubkey].elapsedNow()`, which was still the giant pre-recycle value (now > 12 s). The check `>= 4_000ms` immediately tripped, recycling AGAIN regardless of whether the new session had begun delivering frames. The detector treated the recycle moment as if no time had passed. 2. The 8 s cooldown was the bare minimum needed for one reconnect handshake. It did NOT include time for moq-rs to drain its per-subscriber forward task queue from the prior subscription. Aggressive recycles every ~12 s drove the relay into a state where it FIN'd new subscribe bidis without responding — visible as "NestsListener.subscribe requires Connected state, was Closed" loops on the receiver after the 4th cliff. Two changes: a) When the cliff detector fires, before calling `recycleSession()` it now writes `lastFrameAt[pubkey] = recycleMark` for every stalled pubkey. This treats the recycle as a synthetic frame event so the cliff timer restarts from the recycle moment. The new session has the full `ROOM_AUDIO_CLIFF_TIMEOUT_MS` (4 s) to deliver before the next cliff check trips, instead of tripping the moment the cooldown ends. b) `ROOM_AUDIO_CLIFF_COOLDOWN_MS: 8_000L → 30_000L`. Gives moq-rs enough wall-clock between recycles to drain its per-subscriber forward queue before the next subscribe lands. Audible-gap cost rises from ~5 s to ~30 s in the worst-case fully-stalled relay, but this is the correct trade: 30 s of silence followed by recovered audio beats endless silence with the relay locked out. Combined, the worst-case recycle cycle goes from ~12 s (8 s cooldown + 4 s threshold = next recycle, relay degrades fast) to ~34 s (30 s cooldown + 4 s threshold = next recycle, relay can recover). For the steady state where the relay is healthy and the recycle DOES help, the new session's first frames clear the threshold on `lastFrameAt` and no second recycle fires at all. Tests: `CliffDetectorTest`'s 12 pure-predicate cases stay green (updated `cooldownSuppressesRecycleEvenWhenStalled` and `cooldownReleasesAfterTimeoutPasses` to use the new 30 s default). The `lastFrameAt` reset on recycle is in the live detector loop, not the predicate, so it's covered by integration runs rather than the pure-function suite. --- .../commons/viewmodels/NestViewModel.kt | 46 +++++++++++++++---- .../commons/viewmodels/CliffDetectorTest.kt | 10 ++-- 2 files changed, 44 insertions(+), 12 deletions(-) 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 c221bdc20..8dd9b6f41 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 @@ -1396,9 +1396,31 @@ class NestViewModel( 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 = + val recycleMark = kotlin.time.TimeSource.Monotonic .markNow() + lastCliffRecycleAt = recycleMark + // Reset `lastFrameAt` for every stalled pubkey so + // the cliff timer starts counting from the recycle + // moment, not from the old (pre-recycle) last + // frame timestamp. Without this reset the next + // tick after the cooldown immediately re-trips — + // because `lastFrameAt[pubkey].elapsedNow()` is + // still the giant pre-recycle value plus the + // cooldown window. Production logs at commit + // ea08c43 showed exactly this: 4 recycles in + // ~30 s eventually drove the relay into a + // "subscribe stream FIN before reply" loop where + // it refused fresh subscribes entirely. Using + // the recycle moment as a synthetic frame event + // gives the new session [ROOM_AUDIO_CLIFF_TIMEOUT_MS] + // wall-clock to deliver before the next cliff + // check, matching the expectation a freshly- + // attached subscription should clear inside + // that window if the relay is healthy. + for (pubkey in stalled) { + lastFrameAt[pubkey] = recycleMark + } runCatching { l.recycleSession() } } } finally { @@ -1698,14 +1720,22 @@ const val ROOM_AUDIO_CLIFF_CHECK_INTERVAL_MS: Long = 1_000L /** * Cooldown after a cliff-detector-driven recycle. Suppresses * back-to-back recycles while the wrapper is mid-handshake on the - * fresh session — a brand-new session has no `lastFrameAt` entries - * yet, which would otherwise immediately re-trigger the detector - * on the next 1 s tick. Long enough to cover the typical reconnect - * handshake plus a couple of audio-frame round trips, short enough - * that a recycle that didn't actually clear the cliff is recovered - * from quickly. + * fresh session AND while the relay is recovering its + * per-subscriber forward queue. + * + * Production logs at commit ea08c43 showed an 8 s cooldown was + * not enough — moq-rs needed longer to recover between aggressive + * recycles, and 4 recycles in ~30 s drove the relay into a + * "subscribe stream FIN before reply" loop that refused all + * subsequent subscribes. 30 s gives the relay's per-subscriber + * forward task pool time to drain stale pending writes from the + * previous subscription before the new subscribe lands. The + * audible-gap cost rises from ~5 s to ~30 s in the worst case + * (a fully-stalled relay), but the trade is correct: 30 s of + * silence + recovered audio beats endless silence with the relay + * locked out. */ -const val ROOM_AUDIO_CLIFF_COOLDOWN_MS: Long = 8_000L +const val ROOM_AUDIO_CLIFF_COOLDOWN_MS: Long = 30_000L /** * Diagnostic log frequency for the cliff detector. Emit a state-dump 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 index 4b8f285c6..a4cc54df2 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/CliffDetectorTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/CliffDetectorTest.kt @@ -193,13 +193,15 @@ class CliffDetectorTest { // 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. + // hasn't been updated by the new session yet). 30 s cooldown + // covers the typical reconnect handshake AND gives moq-rs + // time to drain its per-subscriber forward queue from the + // prior subscription before the new subscribe lands. 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 + ts += 5_000.milliseconds // 5 s into cooldown — well within 30 s window val result = computeStalledSpeakers( @@ -221,7 +223,7 @@ class CliffDetectorTest { val frameMark = ts.markNow() ts += 4_500.milliseconds val recycleMark = ts.markNow() - ts += 8_001.milliseconds // 1 ms past cooldown + ts += 30_001.milliseconds // 1 ms past 30 s cooldown val result = computeStalledSpeakers(