diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/ParticipantsGrid.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/ParticipantsGrid.kt index 0091834e1..d79d7eb29 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/ParticipantsGrid.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/ParticipantsGrid.kt @@ -57,6 +57,7 @@ import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.draw.alpha import androidx.compose.ui.draw.drawBehind +import androidx.compose.ui.geometry.Offset import androidx.compose.ui.graphics.Color import androidx.compose.ui.graphics.drawscope.Stroke import androidx.compose.ui.res.pluralStringResource @@ -420,28 +421,18 @@ private fun MemberCell( }, label = "speaker-outer-ring-width", ) + // Reserve enough space around the avatar Box to fit the outer ring + // and glow halo. Without this padding the rings clip against the + // surrounding Surface / LazyVerticalGrid bounds (most visibly at + // the top edge for the first row, where the stage card's rounded + // corner cuts into the glow). The glow extends up to MAX_GLOW_RADIUS + // past the avatar; the outer ring extends OUTER_RING_GAP + + // OUTER_RING_MAX_WIDTH past it. + val ringPadding = + maxOf(MAX_GLOW_RADIUS.value, (OUTER_RING_GAP + OUTER_RING_MAX_WIDTH).value).dp val avatarModifier = Modifier - .drawBehind { - if (animatedGlowAlpha > 0.001f) { - val baseRadius = size.minDimension / 2f - val extra = MAX_GLOW_RADIUS.toPx() * clampedLevel - drawCircle( - color = NEST_SPEAKING_COLOR.copy(alpha = animatedGlowAlpha), - radius = baseRadius + extra, - ) - } - if (animatedOuterRingAlpha > 0.001f && animatedOuterRingWidth > 0.dp) { - val baseRadius = size.minDimension / 2f - val strokePx = animatedOuterRingWidth.toPx() - val ringRadius = baseRadius + OUTER_RING_GAP.toPx() + strokePx / 2f - drawCircle( - color = NEST_SPEAKING_COLOR.copy(alpha = animatedOuterRingAlpha), - radius = ringRadius, - style = Stroke(width = strokePx), - ) - } - }.border(animatedRingWidth, animatedRingColor, CircleShape) + .border(animatedRingWidth, animatedRingColor, CircleShape) .let { if (member.absent) it.alpha(0.5f) else it } val user = remember(member.pubkey) { @@ -480,58 +471,53 @@ private fun MemberCell( horizontalAlignment = Alignment.CenterHorizontally, modifier = modifier.fillMaxWidth().padding(vertical = 4.dp), ) { - Box(contentAlignment = Alignment.Center) { - ClickableUserPicture( - baseUserHex = member.pubkey, - size = avatarSize, - accountViewModel = accountViewModel, - modifier = avatarModifier, - onClick = onClick, - onLongClick = onLongClick, - ) - if (isConnecting) { - CircularProgressIndicator( - modifier = Modifier.size(avatarSize - 8.dp), - strokeWidth = 2.dp, - color = MaterialTheme.colorScheme.primary, - ) - } - val role = member.role - if (role == ROLE.HOST || role == ROLE.MODERATOR) { - RoleBadge( - role = role, - modifier = Modifier.align(Alignment.TopStart), - ) - } - if (member.handRaised) { - HandRaiseBadge( - modifier = Modifier.align(Alignment.TopEnd), - ) - } - // Show the mic badge for any on-stage speaker that has - // an audio state to surface ā€” currently broadcasting - // (`publishing=1`) OR mic-muted (`muted=1, publishing=0`). - // Gating only on `publishing` would hide the muted icon - // the moment the user mutes, which is exactly when it's - // supposed to appear. - if (showMicBadge && (member.publishing || member.muted == true)) { - MicStateBadge( + // Outer Box paints the glow halo + detached outer ring on a + // canvas that's bigger than the avatar by [ringPadding]. The + // inner Box keeps its tight-to-avatar bounds so badge corner + // alignment (TopStart, TopEnd, BottomCenter, BottomEnd) still + // tracks the avatar circle, not the padded outer area. + Box( + modifier = + Modifier.drawBehind { + val avatarRadiusPx = avatarSize.toPx() / 2f + val cx = size.width / 2f + val cy = size.height / 2f + if (animatedGlowAlpha > 0.001f) { + val extra = MAX_GLOW_RADIUS.toPx() * clampedLevel + drawCircle( + color = NEST_SPEAKING_COLOR.copy(alpha = animatedGlowAlpha), + radius = avatarRadiusPx + extra, + center = Offset(cx, cy), + ) + } + if (animatedOuterRingAlpha > 0.001f && animatedOuterRingWidth > 0.dp) { + val strokePx = animatedOuterRingWidth.toPx() + val ringRadius = avatarRadiusPx + OUTER_RING_GAP.toPx() + strokePx / 2f + drawCircle( + color = NEST_SPEAKING_COLOR.copy(alpha = animatedOuterRingAlpha), + radius = ringRadius, + center = Offset(cx, cy), + style = Stroke(width = strokePx), + ) + } + }, + contentAlignment = Alignment.Center, + ) { + Box( + modifier = Modifier.padding(ringPadding), + contentAlignment = Alignment.Center, + ) { + AvatarAndBadges( + member = member, + avatarSize = avatarSize, + accountViewModel = accountViewModel, + avatarModifier = avatarModifier, + onClick = onClick, + onLongClick = onLongClick, + isConnecting = isConnecting, + showMicBadge = showMicBadge, isSpeaking = isSpeaking, - isMuted = member.muted == true, - modifier = Modifier.align(Alignment.BottomCenter), - ) - } - // Reactions float over the avatar's bottom-right corner so - // a šŸ‘ burst no longer pushes the username down and reflows - // neighbouring cells. The mic badge sits at BottomCenter, - // so BottomEnd + a small outward offset keeps them clear. - if (reactions.isNotEmpty()) { - SpeakerReactionOverlay( reactions = reactions, - modifier = - Modifier - .align(Alignment.BottomEnd) - .offset(x = 6.dp, y = 6.dp), ) } } @@ -545,6 +531,76 @@ private fun MemberCell( } } +@Composable +private fun AvatarAndBadges( + member: RoomMember, + avatarSize: Dp, + accountViewModel: AccountViewModel, + avatarModifier: Modifier, + onClick: ((String) -> Unit)?, + onLongClick: ((String) -> Unit)?, + isConnecting: Boolean, + showMicBadge: Boolean, + isSpeaking: Boolean, + reactions: List, +) { + Box(contentAlignment = Alignment.Center) { + ClickableUserPicture( + baseUserHex = member.pubkey, + size = avatarSize, + accountViewModel = accountViewModel, + modifier = avatarModifier, + onClick = onClick, + onLongClick = onLongClick, + ) + if (isConnecting) { + CircularProgressIndicator( + modifier = Modifier.size(avatarSize - 8.dp), + strokeWidth = 2.dp, + color = MaterialTheme.colorScheme.primary, + ) + } + val role = member.role + if (role == ROLE.HOST || role == ROLE.MODERATOR) { + RoleBadge( + role = role, + modifier = Modifier.align(Alignment.TopStart), + ) + } + if (member.handRaised) { + HandRaiseBadge( + modifier = Modifier.align(Alignment.TopEnd), + ) + } + // Show the mic badge for any on-stage speaker that has + // an audio state to surface ā€” currently broadcasting + // (`publishing=1`) OR mic-muted (`muted=1, publishing=0`). + // Gating only on `publishing` would hide the muted icon + // the moment the user mutes, which is exactly when it's + // supposed to appear. + if (showMicBadge && (member.publishing || member.muted == true)) { + MicStateBadge( + isSpeaking = isSpeaking, + isMuted = member.muted == true, + modifier = Modifier.align(Alignment.BottomCenter), + ) + } + // Reactions float over the avatar's bottom-right corner so + // a šŸ‘ burst no longer pushes the username down and reflows + // neighbouring cells. The mic badge sits at BottomCenter, + // so BottomEnd + a small outward offset keeps them clear. + if (reactions.isNotEmpty()) { + SpeakerReactionOverlay( + reactions = reactions, + modifier = + Modifier + .align(Alignment.BottomEnd) + .offset(x = 6.dp, y = 6.dp), + ) + } + } +} + /** * Hand-raise indicator overlaid on the avatar ā€” yellow circle with * a hand glyph at the top-right, animated in a subtle vertical 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..2fd58dfa0 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 @@ -1469,9 +1494,17 @@ class NestViewModel( } /** - * Mark [pubkey] as currently speaking and (re)arm a [SPEAKING_TIMEOUT_MS] - * coroutine that clears it once they go quiet. Called once per - * MoQ object received on the speaker's track. + * Per-frame heartbeat. Called once per MoQ object received on the + * speaker's track ā€” i.e. once per ~20 ms regardless of whether + * that frame contains actual speech, silence, or background noise. + * + * Bumps the cliff-detector timestamp so an active stream keeps + * resetting the relay-forward-queue stall watchdog, and clears + * the per-speaker buffering overlay the first time a frame lands. + * + * NB: this does NOT mark the speaker as "speaking right now". + * That signal is energy-gated and lives in [onAudioLevel] ā€” + * mic-on with no voice MUST NOT light up the green ring. */ private fun onSpeakerActivity(pubkey: String) { if (closed) return @@ -1483,12 +1516,27 @@ class NestViewModel( lastFrameAt[pubkey] = kotlin.time.TimeSource.Monotonic .markNow() - speakingExpiryJobs[pubkey]?.cancel() + // 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 // First frame for this subscription ā€” clear the buffering // overlay. Subsequent frames are no-ops here. if (_uiState.value.connectingSpeakers.contains(pubkey)) { _uiState.update { it.copy(connectingSpeakers = (it.connectingSpeakers - pubkey).toPersistentSet()) } } + } + + /** + * Mark [pubkey] as currently speaking and (re)arm a + * [SPEAKING_TIMEOUT_MS] coroutine that clears the flag once their + * audio drops back below the threshold for that long. Called from + * [onAudioLevel] only when the decoded peak is loud enough to + * read as speech (see [SPEAKING_LEVEL_THRESHOLD]). + */ + private fun markSpeaking(pubkey: String) { + speakingExpiryJobs[pubkey]?.cancel() if (!_uiState.value.speakingNow.contains(pubkey)) { _uiState.update { it.copy(speakingNow = (it.speakingNow + pubkey).toPersistentSet()) } } @@ -1570,43 +1618,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 @@ -1641,6 +1699,20 @@ class NestViewModel( ) { if (closed) return rawAudioLevels[pubkey] = level + // Energy-gated speaking detector. The MoQ track delivers a + // frame every ~20 ms while the mic is open, even when the + // speaker is silent or only picking up room noise ā€” gating the + // green ring on "frame arrived" therefore lights it up the + // moment the mic is unmuted, not when there's actually a voice + // on it. The decoded peak amplitude (`peakAmplitude` in + // nestsClient/audio/Amplitude.kt) gives us the signal we need: + // background noise / breath stays under a few percent of full + // scale, while even a quiet voice clears [SPEAKING_LEVEL_THRESHOLD]. + // The 250 ms expiry already wired up in [markSpeaking] gives + // the indicator natural hysteresis between syllables. + if (level >= SPEAKING_LEVEL_THRESHOLD) { + markSpeaking(pubkey) + } startLevelEmitter() } @@ -1799,12 +1871,29 @@ sealed class BroadcastUiState { } /** - * How long a speaker stays "speaking" after their last received MoQ object. - * Roughly 12 Ɨ the 20 ms Opus frame so brief packet jitter doesn't make the - * indicator flicker. + * How long a speaker stays "speaking" after their last decoded frame + * over the [SPEAKING_LEVEL_THRESHOLD]. Roughly 12 Ɨ the 20 ms Opus + * frame so brief packet jitter and inter-syllable pauses don't make + * the indicator flicker between adjacent words. */ const val SPEAKING_TIMEOUT_MS: Long = 250L +/** + * Minimum decoded peak amplitude (normalized to `[0, 1]`) that counts + * as "this person is actually speaking right now". Frames whose peak + * lands below this threshold are treated as silence / room tone / + * breath ā€” they keep the per-speaker subscription healthy (the cliff + * heartbeat in [NestViewModel.onSpeakerActivity] still fires) but do + * NOT light up the green speaking ring. + * + * 0.06 ā‰ˆ -24 dBFS, comfortably above the typical residential-mic + * noise floor (~-40 to -30 dBFS) while still tripping on a quiet + * voice. Tuned in conjunction with [SPEAKING_TIMEOUT_MS]: a single + * loud frame is enough to arm the indicator; ā‰„ 250 ms below the + * threshold drops it. + */ +const val SPEAKING_LEVEL_THRESHOLD: Float = 0.06f + /** * How long [NestViewModel.openSubscription] waits for the publisher's * `catalog.json` to land before constructing the decoder + AudioTrack. @@ -1926,24 +2015,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 +2053,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 +2103,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 diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt index 5c7415ab5..59b7a91ca 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt @@ -55,6 +55,7 @@ import kotlin.test.assertEquals import kotlin.test.assertFalse import kotlin.test.assertIs import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.seconds /** * Drives [NestViewModel] with a fake [NestsListenerConnector] and @@ -72,6 +73,16 @@ import kotlin.test.assertTrue */ @OptIn(ExperimentalCoroutinesApi::class) class NestViewModelTest { + // Tracks every VM created by `newViewModel` so each test can dispose them + // before runTest exits. `connect()` starts an infinite cliff-detector + // loop in viewModelScope; without an explicit teardown that loop spins + // forever under runTest's virtual scheduler and the test wedges until + // the real-time runTest deadline (60 s Ɨ N tests => :commons:jvmTest + // hangs). [runVmTest] wraps each test body with a finally that calls + // `disconnect()` (which cancels cliffDetectorJob via teardown()) so the + // scheduler can become idle and runTest can return. + private val createdVms = mutableListOf() + @BeforeTest fun setupMainDispatcher() { Dispatchers.setMain(UnconfinedTestDispatcher()) @@ -79,12 +90,34 @@ class NestViewModelTest { @AfterTest fun resetMainDispatcher() { + // Belt-and-braces: if a test crashes before its finally runs, this + // still tears down so the next test starts clean. + createdVms.forEach { runCatching { it.disconnect() } } + createdVms.clear() Dispatchers.resetMain() } + /** + * Test wrapper that guarantees every VM created via [newViewModel] is + * disconnected before [runTest] tries to drain the test scheduler. The + * 10-second [runTest] timeout is a safety net ā€” a healthy test in this + * class finishes in milliseconds; if we trip the timeout it almost + * certainly means a new code path is leaving viewModelScope coroutines + * running and needs its own teardown call. + */ + private fun runVmTest(body: suspend TestScope.() -> Unit) = + runTest(timeout = 10.seconds) { + try { + body() + } finally { + createdVms.forEach { runCatching { it.disconnect() } } + createdVms.clear() + } + } + @Test fun connectShowsConnectingThenConnected() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() val vm = newViewModel { fakeListener } @@ -103,7 +136,7 @@ class NestViewModelTest { @Test fun listenerFailedSurfacesAsUiFailed() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() val vm = newViewModel { fakeListener } @@ -117,7 +150,7 @@ class NestViewModelTest { @Test fun connectorThrowsBecomesUiFailed() = - runTest { + runVmTest { val vm = newViewModel { throw NestsException("dns blew up") } vm.connect() @@ -129,7 +162,7 @@ class NestViewModelTest { @Test fun setMutedFlipsUiStateAndIsRetained() = - runTest { + runVmTest { val vm = newViewModel { FakeNestsListener() } assertFalse(vm.uiState.value.isMuted) @@ -141,7 +174,7 @@ class NestViewModelTest { @Test fun onStageNowDefaultsTrueAndSetOnStageFlipsIt() = - runTest { + runVmTest { val vm = newViewModel { FakeNestsListener() } // Defaults to true so a freshly-joined speaker advertises @@ -156,7 +189,7 @@ class NestViewModelTest { @Test fun onPresenceEventPopulatesPresencesMapAndDedupesByPubkey() = - runTest { + runVmTest { val vm = newViewModel { FakeNestsListener() } val alice = "a".repeat(64) @@ -187,7 +220,7 @@ class NestViewModelTest { @Test fun evictStalePresencesDropsOldPeers() = - runTest { + runVmTest { val vm = newViewModel { FakeNestsListener() } val alice = "a".repeat(64) val bob = "b".repeat(64) @@ -219,7 +252,7 @@ class NestViewModelTest { @Test fun onChatEventAccumulatesMessagesSortedByCreatedAt() = - runTest { + runVmTest { val vm = newViewModel { FakeNestsListener() } val alice = "a".repeat(64) @@ -255,7 +288,7 @@ class NestViewModelTest { @Test fun onChatEventDedupesByEventId() = - runTest { + runVmTest { val vm = newViewModel { FakeNestsListener() } val alice = "a".repeat(64) val msg = @@ -281,7 +314,7 @@ class NestViewModelTest { @Test fun onReactionEventGroupsByTargetAndEvictsOnTick() = - runTest { + runVmTest { val vm = newViewModel { FakeNestsListener() } val alice = "a".repeat(64) val bob = "b".repeat(64) @@ -326,7 +359,7 @@ class NestViewModelTest { @Test fun onKickFlipsWasKickedAndDisconnects() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() val vm = newViewModel { fakeListener } vm.connect() @@ -343,7 +376,7 @@ class NestViewModelTest { @Test fun onKickIsIdempotent() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() val vm = newViewModel { fakeListener } vm.connect() @@ -377,7 +410,7 @@ class NestViewModelTest { @Test fun connectIsIdempotentWhileConnecting() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() var connectCalls = 0 val vm = @@ -395,7 +428,7 @@ class NestViewModelTest { @Test fun disconnectReturnsToIdleAndClosesListener() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() val vm = newViewModel { fakeListener } @@ -411,7 +444,7 @@ class NestViewModelTest { @Test fun connectingStepMapsThroughToUiStep() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() val vm = newViewModel { fakeListener } @@ -429,7 +462,7 @@ class NestViewModelTest { @Test fun speakingNowClearsOnTeardown() = - runTest { + runVmTest { val fakeListener = FakeNestsListener() val vm = newViewModel { fakeListener } @@ -466,7 +499,7 @@ class NestViewModelTest { // Wire to the test's backgroundScope so close calls run during // the test rather than escaping to the real GlobalScope. cleanupScope = backgroundScope, - ) + ).also { createdVms.add(it) } private class FakeNestsListener : NestsListener { private val mutable = MutableStateFlow(NestsListenerState.Idle) diff --git a/nestsClient/plans/2026-05-06-stream-priority-followup.md b/nestsClient/plans/2026-05-06-stream-priority-followup.md deleted file mode 100644 index 6e9578214..000000000 --- a/nestsClient/plans/2026-05-06-stream-priority-followup.md +++ /dev/null @@ -1,151 +0,0 @@ -# Stream priority for moq-lite group uni streams (T11.3 follow-up) - -**Status**: deferred ā€” flagged in T11 (drop `bestEffort=true`), not landed. - -## Why - -`T11` removed `bestEffort=true` from `MoqLiteSession.openGroupStream` -because the QUIC contract it relied on (drop lost ranges silently -without `RESET_STREAM`) created undeliverable streams that wedge -peer reassembly buffers ā€” the actual user-visible bug was a 30 s -silent dropout per lost packet on lossy networks. - -`bestEffort=true` was incidentally providing a "newer-groups-skip- -queued-retransmits" effect: the writer never queued retransmits in -the first place, so under congestion the loss budget naturally -biased toward dropping older lost ranges. With `bestEffort=true` -gone, all retransmits are now queued, and under sustained -congestion the writer drains streams in `streamRoundRobinStart` -order ā€” which can mean the listener catches up on a stale group -when a fresh one would be more useful. - -The kixelated reference (`moq-rs`'s `Publisher::serve_group` in -`rs/moq-lite/src/lite/publisher.rs:347-406`) addresses this by -calling `stream.set_priority(priority.current())` on every group -stream and biasing the writer toward higher-priority (newer) -streams. We should do the same. - -## Target shape - -### `:quic` API - -Add a stable priority hook to `QuicStream`: - -```kotlin -class QuicStream(...) { - @Volatile - var priority: Int = 0 - // Higher drains first under contention. Default 0 = unchanged - // round-robin behaviour. -} -``` - -Expose at the WebTransport layer: - -```kotlin -interface WebTransportWriteStream { - fun setPriority(priority: Int) -} -``` - -### `QuicConnectionWriter` ā€” the load-bearing change - -Today's send-frame loop (`QuicConnectionWriter.kt:411-416`): - -```kotlin -val streamsView = conn.streamsListLocked() -val start = conn.streamRoundRobinStart % streamsView.size -for (i in streamsView.indices) { - val stream = streamsView[(start + i) % streamsView.size] - // ... -} -``` - -Replace with priority-then-round-robin: - -```kotlin -val streamsView = conn.streamsListLocked() -val sorted = streamsView.sortedByDescending { it.priority } -val start = conn.streamRoundRobinStart % sorted.size -for (i in sorted.indices) { - val stream = sorted[(start + i) % sorted.size] - // ... -} -``` - -Sort is stable; same-priority streams retain insertion order, so the -existing round-robin behaviour holds within a priority tier. Higher- -priority streams always come first in the iteration. - -Cost: O(N log N) per drain pass, where N is active local-initiated -streams. N is small (1ā€“10 in the moq-lite audio path); the -allocation of `sorted` per pass is the only real cost. If that -shows up in profiling, switch to `kotlin.collections.IntArray`- -backed indirect sort or maintain a priority-sorted list incrementally -on `setPriority` calls. - -### moq-lite wiring - -`MoqLiteSession.openGroupStream:1022-1037`: - -```kotlin -internal suspend fun openGroupStream( - subscribeId: Long, - sequence: Long, -): WebTransportWriteStream { - val uni = transport.openUniStream() - uni.setPriority(sequence.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) - uni.write(Varint.encode(MoqLiteDataType.Group.code)) - uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence))) - return uni -} -``` - -Newer groups have higher sequence ā†’ higher priority ā†’ drain first -under congestion. The saturation conversion handles broadcasts that -run long enough for `sequence` to exceed `Int.MAX_VALUE` (ā‰ˆ 71 -years at our 1 group/sec production cadence; defensive only). - -## Test - -Add a unit test in `:quic` that builds a `QuicConnection`, opens -two streams, sets `.priority = 0` on the first and `.priority = 1` -on the second, queues bytes on both, and verifies the higher- -priority stream's bytes hit the wire first. Doesn't need to fake -real flow-control backpressure ā€” pinning the iteration order via -the writer's emitted-frames tape is sufficient to catch the -regression case ("a later refactor accidentally re-introduces -round-robin order"). - -A second test in `:nestsClient` that opens 3 group streams and -asserts later-sequence streams drain before earlier ones is -nice-to-have but secondary; the `:quic`-level test pins the load- -bearing invariant. - -## Why deferred - -The change is small in lines but touches the QUIC writer's hot -path. Rebasing a future `:quic` retransmit / pacing change onto a -priority-sorted iteration order is doable but the diff conflicts -get noisy. Better landed as a focused PR after the current -interop-work series stabilises. - -Risk profile: -- **Bugs**: subtle starvation risk if a high-priority stream - always has `streamRemaining > 0` ā€” round-robin tiebreaker within - a priority tier mitigates this for same-priority case, but the - cross-tier case needs deliberate thought (do we want strict - priority or weighted?). -- **Performance**: per-pass sort allocation. Negligible for Nā‰¤10 - but worth measuring if N grows. -- **Compat**: streams without an explicit priority default to 0, - matching today's behaviour. Existing tests should pass unchanged. - -## When to land - -After the catalog interop series is fully verified (production -audio with no degradation under realistic conditions), and ideally -alongside a `:quic` perf review pass that touches the same code -path. Not blocking on any audio bug today ā€” `T11`'s reliable- -delivery fix is the actual correctness change; this is the -spec-aligned hardening. diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 3dc3825ed..944eb1ae1 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -1037,6 +1037,13 @@ class MoqLiteSession internal constructor( // 150 ms late still falls inside hang's default ~200 ms // jitter buffer) and avoids the dropout entirely. val uni = transport.openUniStream() + // Mirror moq-rs `Publisher::serve_group` + // (`rs/moq-lite/src/lite/publisher.rs`): newer groups get higher + // priority so the QUIC writer drains fresh audio first when + // retransmits queue up under loss. Saturating cast guards a + // theoretical broadcast that runs long enough for `sequence` to + // exceed Int.MAX_VALUE (ā‰ˆ 71 years at 1 group/sec); defensive only. + uni.setPriority(sequence.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) uni.write(Varint.encode(MoqLiteDataType.Group.code)) uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence))) return uni diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt index 4c79793f2..875bcd568 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt @@ -162,6 +162,8 @@ class FakeBidiStream internal constructor( override suspend fun finish() { write.close() } + + override fun setPriority(priority: Int) = Unit } class FakeReadStream internal constructor( @@ -186,4 +188,6 @@ private class ChannelWriteStream( override suspend fun finish() { channel.close() } + + override fun setPriority(priority: Int) = Unit } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt index e0c952f39..f581ea986 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt @@ -112,6 +112,22 @@ interface WebTransportWriteStream { /** Half-close the write side (FIN). No further writes after this call. */ suspend fun finish() + + /** + * Hint to the transport about this stream's drain priority relative + * to other streams on the same session. Higher value drains first + * under congestion; same-priority streams keep round-robin order. + * Default 0 = unchanged round-robin behaviour. + * + * moq-lite uses this to bias the writer toward newer group streams + * (sequence-numbered, fresher audio) so a backlog of retransmits on + * an older group doesn't starve the listener of fresh frames. See + * `Publisher::serve_group` in `rs/moq-lite/src/lite/publisher.rs`. + * + * Implementations that don't model priority (e.g. the in-memory + * fake) MAY treat this as a no-op. + */ + fun setPriority(priority: Int) } /** diff --git a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt index 4f5fe7e59..019ef6e21 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt @@ -365,6 +365,10 @@ private class QuicBidiStreamAdapter( stream.send.finish() driver.wakeup() } + + override fun setPriority(priority: Int) { + stream.priority = priority + } } private class QuicReadStreamAdapter( @@ -392,6 +396,10 @@ private class QuicUniWriteStreamAdapter( stream.send.finish() driver.wakeup() } + + override fun setPriority(priority: Int) { + stream.priority = priority + } } /** Adapter for a WT peer-initiated uni stream whose prefix has been stripped. */ @@ -431,4 +439,14 @@ private class StrippedWtBidiStreamAdapter( ?: error("peer-initiated bidi stream has no finish ā€” demux didn't wire one") finish() } + + /** + * No-op: peer-initiated bidi streams arrive through the demux as a + * [com.vitorpamplona.quic.webtransport.StrippedWtStream] which exposes + * only `send`/`finish` closures, not the underlying [QuicStream]. The + * moq-lite priority use case targets locally-opened uni group streams + * only, so this path doesn't need to model priority ā€” see the + * [WebTransportWriteStream.setPriority] contract. + */ + override fun setPriority(priority: Int) = Unit } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 96bd15f8f..7b5598738 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -410,46 +410,74 @@ private fun buildApplicationPacket( // insertion-ordered and stays in sync with the streams map. val streamsView = conn.streamsListLocked() if (streamsView.isNotEmpty()) { - val start = conn.streamRoundRobinStart % streamsView.size - for (i in streamsView.indices) { - if (packetBudget <= 64) break - val stream = streamsView[(start + i) % streamsView.size] - val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L) - // Skip if both stream and connection have no credit; FIN-only - // (zero-byte) chunks may still go through because they don't - // consume credit. - if (streamRemaining <= 0L && connBudget <= 0L && !stream.send.finPending) continue - val effectiveCap = minOf(streamRemaining, connBudget) - val maxBytes = - minOf(packetBudget - 32, effectiveCap.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) - val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue - if (chunk.data.isNotEmpty() || chunk.fin) { - frames += - StreamFrame( - streamId = stream.streamId, - offset = chunk.offset, - data = chunk.data, - fin = chunk.fin, - explicitLength = true, - ) - // Step C of the deferred-follow-ups pass: track this - // STREAM emission so RFC 9002 retransmit can re-queue - // the byte range on loss. SendBuffer.markLost (commit B) - // moves the range from in-flight back to the retransmit - // queue, and the next takeChunk replays it. - tokens += - RecoveryToken.Stream( - streamId = stream.streamId, - offset = chunk.offset, - length = chunk.data.size.toLong(), - fin = chunk.fin, - ) - packetBudget -= chunk.data.size + 32 - connBudget -= chunk.data.size - conn.sendConnectionFlowConsumed += chunk.data.size + // Strict priority across tiers, round-robin within each tier. + // Higher-priority streams (e.g. moq-lite newer-sequence group + // streams) ALWAYS drain ahead of lower-priority ones; the + // rotating start-index only rotates among same-priority peers. + // This is the spec-aligned shape ā€” applying the rotation + // globally over the sorted list would flip cross-tier order on + // alternating drains, defeating the priority hint entirely. + // + // Default priority is 0; if every stream is at the default, all + // streams form a single tier and iteration order matches the + // pre-priority round-robin behaviour exactly. + // + // Cost: O(N log N) per drain pass plus one transient sorted + // list. N is small (1ā€“10 in the moq-lite audio path); if it + // ever grows enough to matter, switch to an indirect index sort + // or maintain an incrementally-sorted view on setPriority. + val sorted = + if (streamsView.size > 1) streamsView.sortedByDescending { it.priority } else streamsView + val rotation = conn.streamRoundRobinStart + var tierStart = 0 + outer@ while (tierStart < sorted.size) { + // Walk the contiguous run of same-priority streams. + val tierPriority = sorted[tierStart].priority + var tierEnd = tierStart + 1 + while (tierEnd < sorted.size && sorted[tierEnd].priority == tierPriority) tierEnd++ + val tierSize = tierEnd - tierStart + val tierRotation = if (tierSize > 1) rotation % tierSize else 0 + for (k in 0 until tierSize) { + if (packetBudget <= 64) break@outer + val stream = sorted[tierStart + ((tierRotation + k) % tierSize)] + val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L) + // Skip if both stream and connection have no credit; FIN-only + // (zero-byte) chunks may still go through because they don't + // consume credit. + if (streamRemaining <= 0L && connBudget <= 0L && !stream.send.finPending) continue + val effectiveCap = minOf(streamRemaining, connBudget) + val maxBytes = + minOf(packetBudget - 32, effectiveCap.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) + val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue + if (chunk.data.isNotEmpty() || chunk.fin) { + frames += + StreamFrame( + streamId = stream.streamId, + offset = chunk.offset, + data = chunk.data, + fin = chunk.fin, + explicitLength = true, + ) + // Step C of the deferred-follow-ups pass: track this + // STREAM emission so RFC 9002 retransmit can re-queue + // the byte range on loss. SendBuffer.markLost (commit B) + // moves the range from in-flight back to the retransmit + // queue, and the next takeChunk replays it. + tokens += + RecoveryToken.Stream( + streamId = stream.streamId, + offset = chunk.offset, + length = chunk.data.size.toLong(), + fin = chunk.fin, + ) + packetBudget -= chunk.data.size + 32 + connBudget -= chunk.data.size + conn.sendConnectionFlowConsumed += chunk.data.size + } } + tierStart = tierEnd } - conn.streamRoundRobinStart = (start + 1) % streamsView.size + conn.streamRoundRobinStart = (rotation + 1) % streamsView.size } if (frames.isEmpty()) return null diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index b77095970..05317cb3b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -47,6 +47,27 @@ class QuicStream( val send = SendBuffer(bestEffort = bestEffort) val receive = ReceiveBuffer() + /** + * Send-side scheduling priority. The connection writer's drain loop + * iterates streams by descending priority; same-priority streams keep + * their existing round-robin order. Higher value = drains first under + * congestion. Default 0 matches pre-priority round-robin behaviour for + * every existing call site. + * + * Used by moq-lite group streams: the publisher assigns each new group + * a priority equal to its sequence number so that newer groups + * (fresher audio) drain ahead of older ones when retransmits queue up + * on a lossy link. Mirrors `Publisher::serve_group` in + * `rs/moq-lite/src/lite/publisher.rs` (`stream.set_priority`). + * + * `@Volatile` because callers (e.g. moq-lite's openGroupStream) + * assign from arbitrary coroutines while the writer reads it under + * [com.vitorpamplona.quic.connection.QuicConnection.lock] during a + * drain pass. + */ + @Volatile + var priority: Int = 0 + /** * Bytes received and confirmed contiguous, exposed as a flow to the consumer. * diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt index 80749b9f9..9957825b1 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt @@ -327,6 +327,46 @@ class InMemoryQuicPipe( */ fun buildServerApplicationPacket(frames: List): ByteArray? = buildServerApplicationDatagram(frames) + /** + * Decrypt the application-level (1-RTT) packet inside a client-emitted + * datagram and return its frames in wire order. Test-only helper for + * assertions that depend on per-frame ordering inside a packet (e.g. + * stream priority scheduling). Walks past any coalesced long-header + * packets (Initial / Handshake) at the front of the datagram, since the + * client may still flush ACKs at those levels post-handshake. Returns + * null if no short-header packet is present or decryption fails. + */ + fun decryptClientApplicationFrames(datagram: ByteArray): List? { + if (datagram.isEmpty()) return null + var offset = 0 + while (offset < datagram.size) { + val first = datagram[offset].toInt() and 0xFF + if ((first and 0x80) == 0) { + val proto = serverApplicationRx ?: return null + val parsed = + ShortHeaderPacket.parseAndDecrypt( + bytes = datagram, + offset = offset, + dcidLen = serverScid.length, + aead = proto.aead, + key = proto.key, + iv = proto.iv, + hp = proto.hp, + hpKey = proto.hpKey, + largestReceivedInSpace = applicationPnSpace.largestReceived, + ) ?: return null + applicationPnSpace.observeInbound(parsed.packet.packetNumber, 0L) + return decodeFrames(parsed.packet.payload) + } + // Long header ā€” skip past it using the encoded length field so + // we can inspect any short-header packet that was coalesced + // after it. + val peeked = LongHeaderPacket.peekHeader(datagram, offset) ?: return null + offset += peeked.totalLength + } + return null + } + private fun buildServerInitialPacket(crypto: ByteArray): ByteArray { val proto = PacketProtection( diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt index d08b49ef2..006cda755 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt @@ -110,6 +110,86 @@ class QuicConnectionWriterTest { } } + @Test + fun writer_drains_higher_priority_streams_before_lower_priority() { + // T11.3 follow-up: priority must dominate iteration order on + // EVERY drain, not just the first. The naive "sort by priority + // then apply the existing rotating start globally" shape looks + // right at a glance but the rotating start advances on every + // drain, so cross-tier ordering flips on alternate drains and + // the priority hint is silently defeated. The correct shape is + // strict priority across tiers + round-robin only within a + // tier, which we pin here by draining TWICE and asserting the + // higher-priority stream's StreamFrame lands first in BOTH + // packets. Low-priority stream is opened FIRST so insertion- + // order can't accidentally pass for priority ordering. + runBlocking { + val (client, pipe) = connectedClient() + val low = client.openBidiStream() + val high = client.openBidiStream() + low.priority = 0 + high.priority = 10 + + repeat(2) { round -> + low.send.enqueue(ByteArray(200) { 0xAA.toByte() }) + high.send.enqueue(ByteArray(200) { 0xBB.toByte() }) + + val datagram = drainOutbound(client, nowMillis = 0L) + assertNotNull(datagram, "drain on round $round must emit a packet") + val frames = pipe.decryptClientApplicationFrames(datagram) + assertNotNull(frames, "decrypt must succeed on round $round") + val streamFrames = frames.filterIsInstance() + assertEquals( + 2, + streamFrames.size, + "expected one StreamFrame per stream on round $round, got $streamFrames", + ) + assertEquals( + high.streamId, + streamFrames[0].streamId, + "higher-priority stream must drain first on round $round; " + + "saw ${streamFrames.map { it.streamId }}", + ) + assertEquals(low.streamId, streamFrames[1].streamId, "low second on round $round") + } + } + } + + @Test + fun writer_round_robins_within_a_priority_tier() { + // Regression guard for the tier-local round-robin: same-priority + // streams must still rotate so an early-opened stream doesn't + // monopolise a packet's stream-frame slot indefinitely. We open + // three streams at the default (0) priority and verify the + // rotating start advances by one per drain, matching the + // pre-priority behaviour. + runBlocking { + val (client, pipe) = connectedClient() + val a = client.openBidiStream() + val b = client.openBidiStream() + val c = client.openBidiStream() + // All default priority ā€” single tier, three streams. + val expectedRotation = + listOf( + listOf(a.streamId, b.streamId, c.streamId), + listOf(b.streamId, c.streamId, a.streamId), + listOf(c.streamId, a.streamId, b.streamId), + ) + for ((round, expected) in expectedRotation.withIndex()) { + a.send.enqueue(ByteArray(64) { 0xA1.toByte() }) + b.send.enqueue(ByteArray(64) { 0xB2.toByte() }) + c.send.enqueue(ByteArray(64) { 0xC3.toByte() }) + + val datagram = drainOutbound(client, nowMillis = 0L) + assertNotNull(datagram, "drain $round must emit a packet") + val frames = pipe.decryptClientApplicationFrames(datagram) + assertNotNull(frames, "decrypt must succeed on round $round") + val ids = frames.filterIsInstance().map { it.streamId } + assertEquals(expected, ids, "round $round round-robin order") + } + } + } + @Test fun writer_respects_connection_level_send_credit_cap() { // Audit-4 #9: pre-fix the writer ignored sendConnectionFlowCredit