From 580aaf0201ddb0ce284a03d9d6e63f3de93d5f23 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 6 May 2026 21:26:07 +0000 Subject: [PATCH] fix(nests): replace flat 30s cliff cooldown with consecutive-failed-recycle backoff MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production trace showed a single failed recycle (relay accepted SUBSCRIBE but never opened uni-streams) leaving the user with ~22s of dead air — the 30s cooldown blocked any retry while audio never recovered, then they'd give up. The cooldown couldn't tell "recovered then re-stalled" (safe to wait) from "recycle did nothing" (need to retry sooner). Replace with a per-attempt backoff: 0 → 5s → 12s → 24s → 30s cap. Reset to attempt 0 on any real frame after the recycle, so a recover-then- restall always re-fires immediately. Cumulative wall-clock to 4 recycles goes from "4 in 30s" (the moq-rs wedge case) to "4 in 41s", protecting the relay while letting the typical single-failure case retry inside 5s. Also stop overwriting `lastFrameAt[pubkey]` on recycle: keeping the real last-frame timestamp is what lets the predicate distinguish the two cases via `lastFrameAt > lastRecycleAt`. Hardening on the leave path: - mark `closed` @Volatile so the cliff-detector finally block reliably observes the value set by `leave()` / `onCleared()` (visible in the trace as `cliff-detector EXITED ... closed=false` after a Leave press). - replace `runCatching { recycleSession() }` with explicit `catch (CancellationException)` re-throw so a Leave during recycle exits promptly rather than running one extra loop iteration. https://claude.ai/code/session_015euGg6AERTRGj7HYysKnLV --- .../commons/viewmodels/NestViewModel.kt | 199 +++++++++++++----- .../commons/viewmodels/CliffDetectorTest.kt | 188 +++++++++++++++-- 2 files changed, 312 insertions(+), 75 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 5c85b2735..8ecc9ace6 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 @@ -302,14 +302,26 @@ class NestViewModel( /** * [kotlin.time.TimeMark] of the most recent recycle the cliff - * detector triggered, or `null` if it has never fired. Acts as a - * cooldown so a single cliff event doesn't kick off multiple - * recycles back-to-back while the wrapper is mid-handshake on - * the new session — the new session has no incoming frames yet, - * which would otherwise trip the detector again immediately. + * detector triggered, or `null` if it has never fired. Combined + * with [consecutiveCliffRecycles] to drive the per-attempt + * backoff schedule in [computeStalledSpeakers] — short backoff + * after the first failed recycle so a single moq-rs cliff + * recovers within a few seconds, escalating to the + * [ROOM_AUDIO_CLIFF_BACKOFF_MAX_MS] cap if the relay stays + * stalled across multiple recycles. */ private var lastCliffRecycleAt: kotlin.time.TimeMark? = null + /** + * Number of consecutive `recycleSession()` calls the detector has + * issued without seeing a real frame in between. Reset to 0 in + * [onSpeakerActivity] when any speaker delivers a frame, so a + * recovered-then-restall pattern re-enters with attempt = 0 + * (immediate recycle eligible) rather than inheriting backoff + * from a prior failed-recycle streak. + */ + private var consecutiveCliffRecycles: Int = 0 + /** * Per-speaker catalog-fetch coroutines. Each entry is the * background `subscribeCatalog` collector launched in @@ -320,7 +332,19 @@ class NestViewModel( */ private val catalogJobs = mutableMapOf() private var requestedSpeakers: Set = emptySet() - private var closed = false + + /** + * `@Volatile` because [closed] is read by background coroutines + * (cliff-detector, observe* loops, audio-focus + network observers) + * but written from `leave()` / `onCleared()` which can fire from + * the Activity destroy callback on a different stack frame than + * the coroutine that's about to suspend in [delay] / [collect]. + * Without the marker, a coroutine that resumes from cancellation + * could read a stale `closed = false` even though the user has + * already left the room — visible in the trace as + * `cliff-detector EXITED ... closed=false` after a Leave press. + */ + @Volatile private var closed = false // Speaker / publisher path private var speaker: NestsSpeaker? = null @@ -1389,6 +1413,7 @@ class NestViewModel( cliffDetectorJob = null lastFrameAt.clear() lastCliffRecycleAt = null + consecutiveCliffRecycles = 0 // Audio-focus observation only ends on the final teardown — // a transient disconnect+reconnect (user retry, room swap) // keeps the bus subscription alive so a focus loss that @@ -1483,6 +1508,11 @@ class NestViewModel( lastFrameAt[pubkey] = kotlin.time.TimeSource.Monotonic .markNow() + // A real frame proves the most recent recycle (if any) actually + // fixed the cliff. Reset the consecutive-failed counter so a + // future re-stall starts from attempt 0 (immediate-fire) rather + // than inheriting a long backoff from the prior streak. + if (consecutiveCliffRecycles != 0) consecutiveCliffRecycles = 0 speakingExpiryJobs[pubkey]?.cancel() // First frame for this subscription — clear the buffering // overlay. Subsequent frames are no-ops here. @@ -1570,43 +1600,53 @@ class NestViewModel( announcedSpeakers = announced, lastFrameAt = lastFrameAt, lastRecycleAt = lastCliffRecycleAt, + consecutiveFailedRecycles = consecutiveCliffRecycles, cliffTimeoutMs = ROOM_AUDIO_CLIFF_TIMEOUT_MS, - cooldownMs = ROOM_AUDIO_CLIFF_COOLDOWN_MS, + postRecycleGraceMs = ROOM_AUDIO_CLIFF_RECYCLE_GRACE_MS, ) if (stalled.isEmpty()) continue + consecutiveCliffRecycles += 1 com.vitorpamplona.quartz.utils.Log.w("NestRx") { - "cliff-detector: announced+subscribed but silent for ≥${ROOM_AUDIO_CLIFF_TIMEOUT_MS}ms — recycling session. stalled=$stalled" + "cliff-detector: announced+subscribed but silent for ≥${ROOM_AUDIO_CLIFF_TIMEOUT_MS}ms — recycling session " + + "(consecutive=$consecutiveCliffRecycles). stalled=$stalled" } com.vitorpamplona.nestsclient.trace.NestsTrace.emit("cliff_recycle") { "\"timeout_ms\":$ROOM_AUDIO_CLIFF_TIMEOUT_MS," + + "\"consecutive\":$consecutiveCliffRecycles," + "\"stalled\":${com.vitorpamplona.nestsclient.trace.jsonArrStr(stalled)}" } - val recycleMark = + lastCliffRecycleAt = 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 + // Note: we deliberately do NOT overwrite + // `lastFrameAt[pubkey]` with the recycle moment. + // Keeping the real-frame timestamp lets the next + // tick distinguish "the recycle delivered audio, + // we then re-stalled" (lastFrameAt advances past + // lastCliffRecycleAt — counter resets in + // `onSpeakerActivity`, attempt = 0, fire-eligible + // immediately) from "the recycle did nothing, + // still no frames" (lastFrameAt unchanged, + // counter keeps growing, backoff escalates). + // The previous reset collapsed both cases into + // a flat 30 s wait, which made a single failed + // recycle sound like a 30 s+ dropout to the + // user even when retrying earlier would have + // recovered. + try { + l.recycleSession() + } catch (ce: CancellationException) { + // User left mid-recycle — exit promptly + // rather than letting the next delay() be + // the one to honor the cancel. `runCatching` + // would have swallowed this CE and run one + // more loop body before exiting. + throw ce + } catch (_: Throwable) { + // Any other recycle failure is best-effort; + // keep the loop running so a follow-up tick + // can re-detect the cliff and retry. } - runCatching { l.recycleSession() } } } finally { com.vitorpamplona.quartz.utils.Log @@ -1926,24 +1966,34 @@ const val ROOM_AUDIO_CLIFF_TIMEOUT_MS: Long = 2_500L 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 AND while the relay is recovering its - * per-subscriber forward queue. + * Post-recycle handshake grace. NEVER recycle again within this + * window of the previous recycle — the wrapper is still tearing + * down + reopening the QUIC session, so a "no frames since recycle" + * reading is just the handshake, not a relay-side cliff. * - * 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. + * 3 s covers a typical re-handshake (drain old session UNSUB + + * UNANNOUNCE + WT_CLOSE + open new WebTransport + SUBSCRIBE + + * SUBSCRIBE_OK). Anything shorter risks recycling inside our own + * handshake window; longer wastes audible silence in the recovering- + * within-grace case. */ -const val ROOM_AUDIO_CLIFF_COOLDOWN_MS: Long = 30_000L +const val ROOM_AUDIO_CLIFF_RECYCLE_GRACE_MS: Long = 3_000L + +/** + * Maximum backoff for the consecutive-failed-recycle schedule — + * the cap that [defaultCliffBackoffMs] saturates at. + * + * The earlier flat 30 s cooldown was motivated by production logs + * at commit ea08c43 where 4 back-to-back recycles in ~30 s drove + * moq-rs into a "subscribe stream FIN before reply" loop that + * refused all subsequent subscribes. The replacement schedule + * (5 s → 12 s → 24 s → 30 s, reset on first real frame) keeps + * the same final-state protection — by the 4th consecutive failed + * recycle we're spaced 30 s apart — while letting the *recovering* + * case (your trace: 1st recycle worked, 2nd cliff fires later) + * retry within ~5 s instead of the previous 30 s of dead air. + */ +const val ROOM_AUDIO_CLIFF_BACKOFF_MAX_MS: Long = 30_000L /** * Diagnostic log frequency for the cliff detector. Emit a state-dump @@ -1954,6 +2004,36 @@ const val ROOM_AUDIO_CLIFF_COOLDOWN_MS: Long = 30_000L */ private const val CLIFF_DIAG_LOG_EVERY: Long = 5L +/** + * Default backoff schedule for consecutive failed recycles. Returns + * the minimum elapsed time since [NestViewModel.lastCliffRecycleAt] + * before the [attempt + 1]-th recycle is permitted. + * + * attempt 0 → 0 ms (no recycle yet — first cliff fires immediately) + * attempt 1 → 5 000 ms (one prior recycle, no audio since) + * attempt 2 → 12 000 ms + * attempt 3 → 24 000 ms + * attempt 4+ → 30 000 ms (cap, [ROOM_AUDIO_CLIFF_BACKOFF_MAX_MS]) + * + * Resets to 0 on the first real frame after a recycle (see + * [NestViewModel.onSpeakerActivity]) — so a recover-then-restall + * pattern always gets the immediate-fire treatment again rather + * than inheriting backoff from the prior failed-recycle streak. + * + * Cumulative wall-clock at attempt N: 0 → 5 → 17 → 41 → 71 s. + * 4 consecutive recycles span ~41 s — meaningfully slower than the + * "4 in ~30 s" pattern that wedged moq-rs in commit ea08c43, while + * still letting the typical 1-failure case retry inside 5 s. + */ +internal fun defaultCliffBackoffMs(attempt: Int): Long = + when { + attempt <= 0 -> 0L + attempt == 1 -> 5_000L + attempt == 2 -> 12_000L + attempt == 3 -> 24_000L + else -> ROOM_AUDIO_CLIFF_BACKOFF_MAX_MS + } + /** * Pure logic of the cliff detector — extracted so headless tests can * exercise it with a [kotlin.time.TestTimeSource] without standing up @@ -1974,23 +2054,30 @@ private const val CLIFF_DIAG_LOG_EVERY: Long = 5L * - 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). + * Suppression (when [lastRecycleAt] is non-null): + * - while elapsed since [lastRecycleAt] is below [postRecycleGraceMs], + * never recycle — we're still inside the previous handshake window. + * - while elapsed is below [backoffForAttempt]([consecutiveFailedRecycles]), + * suppress as well: the per-attempt schedule throttles + * consecutive failed recycles so we don't hammer the relay. + * - the consecutive counter is reset by the caller on the first + * real frame after a recycle, so a recovered-then-restall + * case always re-enters with attempt = 0 and fires immediately. */ internal fun computeStalledSpeakers( activeSpeakers: Set, announcedSpeakers: Set, lastFrameAt: Map, lastRecycleAt: kotlin.time.TimeMark?, + consecutiveFailedRecycles: Int = 0, cliffTimeoutMs: Long = ROOM_AUDIO_CLIFF_TIMEOUT_MS, - cooldownMs: Long = ROOM_AUDIO_CLIFF_COOLDOWN_MS, + postRecycleGraceMs: Long = ROOM_AUDIO_CLIFF_RECYCLE_GRACE_MS, + backoffForAttempt: (Int) -> Long = ::defaultCliffBackoffMs, ): List { - if (lastRecycleAt != null && - lastRecycleAt.elapsedNow().inWholeMilliseconds < cooldownMs - ) { - return emptyList() + if (lastRecycleAt != null) { + val sinceRecycleMs = lastRecycleAt.elapsedNow().inWholeMilliseconds + if (sinceRecycleMs < postRecycleGraceMs) return emptyList() + if (sinceRecycleMs < backoffForAttempt(consecutiveFailedRecycles)) return emptyList() } return activeSpeakers .asSequence() 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 ac51d5ce8..3aad186bb 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 @@ -189,21 +189,16 @@ class CliffDetectorTest { } @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). 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. + fun postRecycleGraceSuppressesEvenAtAttemptZero() { + // Inside the 3 s post-recycle handshake window, never recycle + // — the wrapper is still tearing down + reopening QUIC, so + // "no frames since recycle" is just the handshake, not a + // relay-side cliff. val ts = TestTimeSource() val frameMark = ts.markNow() - ts += 4_500.milliseconds // past threshold - val recycleMark = ts.markNow() // recycle just fired - ts += 5_000.milliseconds // 5 s into cooldown — well within 30 s window + ts += 4_500.milliseconds + val recycleMark = ts.markNow() + ts += 2_000.milliseconds // inside 3 s grace val result = computeStalledSpeakers( @@ -211,21 +206,22 @@ class CliffDetectorTest { announcedSpeakers = setOf(ALICE), lastFrameAt = mapOf(ALICE to frameMark), lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 0, ) 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. + fun attemptZeroFiresImmediatelyOnceGracePasses() { + // After a recovered-then-restall pattern: counter has been + // reset by `onSpeakerActivity`, so even though there's a + // prior recycleMark, the next cliff fires as soon as the + // 3 s grace passes — no extended backoff. val ts = TestTimeSource() val frameMark = ts.markNow() ts += 4_500.milliseconds val recycleMark = ts.markNow() - ts += 30_001.milliseconds // 1 ms past 30 s cooldown + ts += 3_500.milliseconds // past 3 s grace, attempt=0 schedule = 0 ms val result = computeStalledSpeakers( @@ -233,10 +229,164 @@ class CliffDetectorTest { announcedSpeakers = setOf(ALICE), lastFrameAt = mapOf(ALICE to frameMark), lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 0, ) assertEquals(listOf(ALICE), result) } + @Test + fun attemptOneBackoffSuppressesUntilFiveSeconds() { + // First failed recycle: schedule says wait 5 s before next. + // 4 s in: still suppressed. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_500.milliseconds + val recycleMark = ts.markNow() + ts += 4_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 1, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun attemptOneBackoffReleasesAtFiveSeconds() { + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_500.milliseconds + val recycleMark = ts.markNow() + ts += 5_000.milliseconds // exactly at attempt-1 boundary + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 1, + ) + assertEquals(listOf(ALICE), result) + } + + @Test + fun attemptTwoBackoffSuppressesUntilTwelveSeconds() { + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_500.milliseconds + val recycleMark = ts.markNow() + ts += 11_000.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 2, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun attemptTwoBackoffReleasesPastTwelveSeconds() { + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_500.milliseconds + val recycleMark = ts.markNow() + ts += 12_500.milliseconds + + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 2, + ) + assertEquals(listOf(ALICE), result) + } + + @Test + fun attemptFourCapsAtThirtySecondMax() { + // Fourth and beyond consecutive failed recycle: backoff + // saturates at the 30 s cap (matching the original flat + // cooldown — by this point we ARE the moq-rs-protection + // case the old constant existed for). + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 4_500.milliseconds + val recycleMark = ts.markNow() + ts += 25_000.milliseconds // past attempt-3 (24 s), inside attempt-4 cap (30 s) + + val resultStillSuppressed = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 4, + ) + assertTrue(resultStillSuppressed.isEmpty()) + + ts += 6_000.milliseconds // now past 30 s cap + val resultReleased = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 4, + ) + assertEquals(listOf(ALICE), resultReleased) + } + + @Test + fun customBackoffFunctionIsHonored() { + // A test can override the schedule (e.g. shorter intervals + // for unit tests that don't want to march through 30 s of + // virtual time) without mutating the production constants. + val ts = TestTimeSource() + val frameMark = ts.markNow() + ts += 3_000.milliseconds + val recycleMark = ts.markNow() + ts += 1_000.milliseconds // past grace=500, past tightBackoff(1)=750 + + val tightBackoff = { attempt: Int -> if (attempt <= 0) 0L else 750L } + val result = + computeStalledSpeakers( + activeSpeakers = setOf(ALICE), + announcedSpeakers = setOf(ALICE), + lastFrameAt = mapOf(ALICE to frameMark), + lastRecycleAt = recycleMark, + consecutiveFailedRecycles = 1, + postRecycleGraceMs = 500L, + backoffForAttempt = tightBackoff, + ) + assertEquals(listOf(ALICE), result) + } + + @Test + fun defaultBackoffSchedulePinsValues() { + // Pin the production schedule so a future tweak is visible + // in code review. attempt 0 → immediate, 1 → 5 s, 2 → 12 s, + // 3 → 24 s, 4+ → 30 s cap. Cumulative wall-clock to the + // Nth recycle: 0, 5, 17, 41, 71 s — slower than the + // 4-recycles-in-30 s pattern that wedged moq-rs in + // commit ea08c43. + assertEquals(0L, defaultCliffBackoffMs(0)) + assertEquals(5_000L, defaultCliffBackoffMs(1)) + assertEquals(12_000L, defaultCliffBackoffMs(2)) + assertEquals(24_000L, defaultCliffBackoffMs(3)) + assertEquals(30_000L, defaultCliffBackoffMs(4)) + assertEquals(30_000L, defaultCliffBackoffMs(10)) + } + @Test fun customTimeoutsAreHonored() { // Defaults are wired into the production VM, but the function