From 8b7785a7e819583c3d9e5487eae9289123c3096c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 1 May 2026 18:41:13 +0000 Subject: [PATCH] fix(nestsClient): token-parse hardening, mint timeout, IPv6 [], broadcaster-bail signal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - NestsTokenResponse.parse catch list now includes IllegalStateException so a malformed escape from a misbehaving auth server surfaces as NestsException instead of crashing the listener. - OkHttpNestsClient gains a configurable callTimeoutMs (default 90 s) enforcing an upper bound on the entire mintToken round-trip including retries. Without it a stalled server suspends the reconnect orchestrator indefinitely (the orchestrator's openOnce step parks on mintToken). 90 s leaves headroom over the worst-case 63 s 429 retry chain documented in MAX_RATE_LIMIT_RETRIES. - parseEndpoint tightens IPv6 bracket check from `closeBracket > 0` to `> 1`, rejecting the empty `[]` literal. - NestBroadcaster + NestMoqLiteBroadcaster gain an `onTerminalFailure` callback that fires once when the consecutive-send-error guard bails. MoqLiteNestsSpeaker wires this to flip the speaker state to Failed, giving ReconnectingNestsSpeaker the signal it needs to recycle the session — without this hook the broadcaster bailed silently and the outward speaker state stayed on Broadcasting forever. Adds regression test: - onTerminalFailure_fires_once_after_consecutive_send_failures 225 tests pass, 0 failures. Android target compiles clean. --- .../nestsclient/MoqLiteNestsSpeaker.kt | 31 +++++++++++- .../vitorpamplona/nestsclient/NestsConnect.kt | 5 +- .../nestsclient/audio/NestBroadcaster.kt | 12 ++++- .../audio/NestMoqLiteBroadcaster.kt | 25 ++++++++-- .../nestsclient/audio/NestBroadcasterTest.kt | 48 +++++++++++++++++++ .../nestsclient/OkHttpNestsClient.kt | 37 +++++++++++++- 6 files changed, 149 insertions(+), 9 deletions(-) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index 2b00d24ed..f3099ba0a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -92,7 +92,20 @@ class MoqLiteNestsSpeaker internal constructor( publisher = publisher, scope = scope, framesPerGroup = framesPerGroup, - ).also { it.start() } + ).also { + it.start( + onTerminalFailure = { + // Broadcaster bailed after sustained + // publisher.send failures. Flip to + // Failed so the reconnect orchestrator + // sees a terminal state and recycles + // the session — without this signal the + // outward state stays on Broadcasting + // and the room is silently mute. + onBroadcastTerminalFailure() + }, + ) + } } catch (t: Throwable) { runCatching { publisher.close() } throw t @@ -129,6 +142,22 @@ class MoqLiteNestsSpeaker internal constructor( } } + /** + * Called from the broadcaster's `onTerminalFailure` callback (off + * the speaker's coroutine). Transitions the speaker to `Failed` so + * the reconnect orchestrator (`ReconnectingNestsSpeaker`) observes + * a terminal state and recycles the session. No-op if the speaker + * is already in a terminal state. + */ + private fun onBroadcastTerminalFailure() { + val current = mutableState.value + if (current is NestsSpeakerState.Failed || current is NestsSpeakerState.Closed) return + mutableState.value = + NestsSpeakerState.Failed( + reason = "broadcast pipeline gave up — likely transport loss", + ) + } + internal fun reportMuteState(muted: Boolean) { val current = mutableState.value if (current is NestsSpeakerState.Broadcasting) { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt index 7c3368c9e..95d7cff70 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt @@ -372,7 +372,10 @@ internal fun parseEndpoint(endpoint: String): Pair { val (hostPart, portStr) = if (authorityRaw.startsWith('[')) { val closeBracket = authorityRaw.indexOf(']') - require(closeBracket > 0) { "malformed IPv6 authority: '$authorityRaw'" } + // closeBracket==1 would mean the literal "[]" — empty IPv6 + // address. Reject so callers can't accidentally pass an + // unconfigured placeholder. + require(closeBracket > 1) { "malformed IPv6 authority: '$authorityRaw'" } val host = authorityRaw.substring(0, closeBracket + 1) val tail = authorityRaw.substring(closeBracket + 1) when { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt index 0d22c062e..d11087329 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt @@ -65,7 +65,16 @@ class NestBroadcaster( * a stopped state and the exception propagates so the caller can * surface it to the user. */ - fun start(onError: (AudioException) -> Unit = { /* swallow */ }) { + fun start( + /** + * See [NestMoqLiteBroadcaster.start]'s `onTerminalFailure` kdoc — + * fires once when the loop bails after MAX_CONSECUTIVE_SEND_ERRORS + * so the speaker can transition to Failed and the orchestrator + * can recycle. + */ + onTerminalFailure: () -> Unit = { /* swallow */ }, + onError: (AudioException) -> Unit = { /* swallow */ }, + ) { check(!stopped) { "NestBroadcaster already stopped" } check(job == null) { "NestBroadcaster.start already called" } @@ -132,6 +141,7 @@ class NestBroadcaster( t, ), ) + runCatching { onTerminalFailure() } return@launch } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index dca21ea3a..0980b3671 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -103,8 +103,20 @@ class NestMoqLiteBroadcaster( * Returns immediately. Calling twice is an error. If * [AudioCapture.start] throws, the broadcaster is left stopped and * the exception propagates. + * + * [onTerminalFailure] fires exactly once when the loop bails after + * [MAX_CONSECUTIVE_SEND_ERRORS] consecutive `publisher.send` + * failures. The broadcaster has stopped capturing by the time this + * callback runs; the caller (typically [MoqLiteNestsSpeaker]) is + * expected to mark the speaker `Failed` so the reconnect + * orchestrator can recycle the session — without this signal the + * outward speaker state stays stuck on `Broadcasting` and the + * orchestrator never knows to act. */ - fun start(onError: (AudioException) -> Unit = { /* swallow */ }) { + fun start( + onTerminalFailure: () -> Unit = { /* swallow */ }, + onError: (AudioException) -> Unit = { /* swallow */ }, + ) { check(!stopped) { "NestMoqLiteBroadcaster already stopped" } check(job == null) { "NestMoqLiteBroadcaster.start already called" } @@ -187,9 +199,14 @@ class NestMoqLiteBroadcaster( t, ), ) - // Falls through to the same post-loop flush path - // capture-EOF takes — caller still owns [stop], - // but the mic-burn loop is over. + // Surface the bail so the speaker + // can flip to Failed and the + // reconnect orchestrator recycles + // the session. Caller still owns + // [stop] — we don't release the mic + // ourselves to avoid double-stop + // races with a concurrent caller. + runCatching { onTerminalFailure() } return@launch } } diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcasterTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcasterTest.kt index d339f5944..e4ef78423 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcasterTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcasterTest.kt @@ -111,6 +111,45 @@ class NestBroadcasterTest { broadcaster.stop() } + @Test + fun onTerminalFailure_fires_once_after_consecutive_send_failures() = + runTest { + // Drive MAX_CONSECUTIVE_SEND_ERRORS + 1 frames through a + // publisher that always throws. The broadcaster must: + // - keep going past one failure (we already cover that) + // - bail eventually + // - fire onTerminalFailure exactly once + // - stop pulling from capture + val frameCount = NestBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS + 50 + val frames = List(frameCount) { shortArrayOf(it.toShort()) } + val capture = ScriptedCapture(frames) + val encoder = ScriptedEncoder(prefix = byteArrayOf(0xFF.toByte())) + val publisher = ThrowingPublisher() + val broadcaster = NestBroadcaster(capture, encoder, publisher, backgroundScope) + val errors = mutableListOf() + var terminalFailureCount = 0 + + broadcaster.start( + onTerminalFailure = { terminalFailureCount += 1 }, + onError = { errors.add(it) }, + ) + + // Wait until the broadcaster has bailed. The bail closes the + // launched job, but capture.awaitDrained only fires on EOF — + // so poll on the terminal-failure counter instead. + while (terminalFailureCount == 0) kotlinx.coroutines.yield() + + assertEquals(1, terminalFailureCount, "onTerminalFailure should fire exactly once") + // We saw at least MAX_CONSECUTIVE_SEND_ERRORS failures before + // the bail, plus one "gave up" message. ScriptedEncoder + // doesn't fail, so all errors are publisher.send failures. + assertTrue( + errors.size >= NestBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS, + "should have seen ≥ ${NestBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS} errors, got ${errors.size}", + ) + broadcaster.stop() + } + @Test fun stop_is_idempotent() = runTest { @@ -207,4 +246,13 @@ class NestBroadcasterTest { closed = true } } + + /** Publisher whose send() always throws — used to drive the bail path. */ + private class ThrowingPublisher : MoqSession.TrackPublisher { + override val name: ByteArray = "throwing".encodeToByteArray() + + override suspend fun send(payload: ByteArray): Boolean = error("publisher.send failure") + + override suspend fun close() {} + } } diff --git a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClient.kt b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClient.kt index 3482f7a2e..ba615e476 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClient.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClient.kt @@ -22,8 +22,10 @@ package com.vitorpamplona.nestsclient import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.delay import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout import okhttp3.MediaType.Companion.toMediaType import okhttp3.OkHttpClient import okhttp3.Request @@ -41,8 +43,16 @@ import kotlin.math.min * OkHttp-backed [NestsClient] used on JVM + Android. A shared [OkHttpClient] * can be injected so the app reuses connection pools / interceptors across * the process; the default constructor creates a dedicated client. + * + * [callTimeoutMs] enforces an upper bound on each `mintToken` round trip, + * including all transport / 429 retries. The injected [OkHttpClient] may + * have its own per-call/connect/read timeouts, but those don't bound the + * retry loop itself — without this watchdog, a stalled server can hold + * the reconnect orchestrator indefinitely (the orchestrator is suspended + * inside `connectNestsListener`'s mint step). */ class OkHttpNestsClient( + private val callTimeoutMs: Long = DEFAULT_CALL_TIMEOUT_MS, private val httpClient: (String) -> OkHttpClient, ) : NestsClient { override suspend fun mintToken( @@ -83,7 +93,17 @@ class OkHttpNestsClient( } return withContext(Dispatchers.IO) { - executeWithRetry(buildRequest, url) + // Hard upper bound on the entire mint round-trip (including + // retries) so a stalled server can't suspend the reconnect + // orchestrator indefinitely. The injected OkHttpClient's + // own callTimeout doesn't cover the retry loop. + val response = + try { + withTimeout(callTimeoutMs) { executeWithRetry(buildRequest, url) } + } catch (e: TimeoutCancellationException) { + throw NestsException("nests mint timed out after ${callTimeoutMs}ms for $url", e) + } + response .use { response -> val body = response.body.string() if (!response.isSuccessful) { @@ -98,6 +118,11 @@ class OkHttpNestsClient( throw NestsException("Malformed nests response from $url", e) } catch (e: IllegalArgumentException) { throw NestsException("Malformed nests response from $url", e) + } catch (e: IllegalStateException) { + // kotlinx.serialization can throw IllegalStateException + // on some malformed input shapes (e.g. unfinished + // escapes) instead of SerializationException. + throw NestsException("Malformed nests response from $url", e) } catch (e: kotlinx.serialization.SerializationException) { throw NestsException("Malformed nests response from $url", e) } @@ -180,10 +205,18 @@ class OkHttpNestsClient( throw NestsException("Failed to reach $url", transportError) } - private companion object { + companion object { private val JSON_MEDIA_TYPE = "application/json".toMediaType() private const val MAX_TRANSPORT_RETRIES = 2 + + /** + * Default upper bound on a full `mintToken` call, including all + * transport / 429 retries. Worst-case 429 backoff totals ~63 s + * per [MAX_RATE_LIMIT_RETRIES] kdoc; 90 s leaves headroom for + * one slow-responding 200 on top of that. + */ + const val DEFAULT_CALL_TIMEOUT_MS: Long = 90_000L } }