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

Merge pull request #2752 from vitorpamplona/claude/fix-amethyst-connection-loss-h34CQ

Browse Source 
Replace cliff-detector cooldown with per-attempt backoff schedule
This commit is contained in:
Vitor Pamplona
2026-05-06 17:34:06 -04:00
committed by GitHub
parent d854d75d7a 580aaf0201
commit d3abf12ac7
2 changed files with 312 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt
+143 -56
View File
@@ -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<String, Job>()
private var requestedSpeakers: Set<String> = 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<String>,
announcedSpeakers: Set<String>,
lastFrameAt: Map<String, kotlin.time.TimeMark>,
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<String> {
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()
commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/CliffDetectorTest.kt
+169 -19
View File
@@ -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