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

fix(nestsClient): token-parse hardening, mint timeout, IPv6 [], broadcaster-bail signal

Browse Source 
- 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.
This commit is contained in:
Claude
2026-05-01 18:41:13 +00:00
parent 702885f4cb
commit 8b7785a7e8
6 changed files with 149 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt
+30 -1
View File
@@ -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) {
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt
+4 -1
View File
@@ -372,7 +372,10 @@ internal fun parseEndpoint(endpoint: String): Pair<String, String> {
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 {
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt
+11 -1
View File
@@ -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
}
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt
+21 -4
View File
@@ -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
}
}
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcasterTest.kt
+48
View File
@@ -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<AudioException>()
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() {}
}
}
nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClient.kt
+35 -2
View File
@@ -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
}
}