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

fix(nestsClient): bug + perf fixes from Nests implementation review

Browse Source 
- send() now nulls currentGroup and returns false on uni-stream write
  failure (was: silently returned true while reusing the dead stream)
- Subscribe bidi gets a single long-running collector that closes the
  frames channel on peer FIN / transport tear-down (was: consumer
  could hang forever on a black-holed UDP path with no Announce(Ended))
- AudioException.Kind gains EncoderError; mic-side encode failures no
  longer mis-route through DecoderError handlers
- MoqCodec.decode bounds the payload-length varint before .toInt() so
  a peer-driven absurd length can't wedge the parser forever
- Publish hot-path framing collapses to a single ByteArray allocation
  (was: Varint.encode + plus operator, two allocs per Opus frame at
  50 fps)
- Drops the now-unused readSubscribeResponseFromBidi /
  readSizePrefixedFromBidiInto / EarlyExit helpers

Adds a regression test that asserts the frames flow completes when
the peer FINs the subscribe bidi.

213 tests pass, 0 failures.
This commit is contained in:
Claude
2026-05-01 17:44:16 +00:00
parent 1a912e5d7c
commit eb20ac2c2a
7 changed files with 146 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusEncoder.kt
+1 -1
View File
@@ -66,7 +66,7 @@ class MediaCodecOpusEncoder(
val inputBuffer =
codec.getInputBuffer(inputIndex)
?: throw AudioException(
AudioException.Kind.DecoderError,
AudioException.Kind.EncoderError,
"MediaCodec returned null input buffer at index $inputIndex",
)
inputBuffer.clear()
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt
+3
View File
@@ -153,6 +153,9 @@ class AudioException(
/** Decoder rejected an Opus packet (corrupted bytes, unsupported config). */
DecoderError,
/** Encoder rejected a PCM frame. */
EncoderError,
/** Audio device resource (AudioTrack/AudioRecord) couldn't be allocated. */
DeviceUnavailable,
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt
+1 -1
View File
@@ -93,7 +93,7 @@ class NestBroadcaster(
} catch (t: Throwable) {
onError(
AudioException(
AudioException.Kind.DecoderError,
AudioException.Kind.EncoderError,
"Opus encode failed for a frame",
t,
),
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt
+1 -1
View File
@@ -134,7 +134,7 @@ class NestMoqLiteBroadcaster(
} catch (t: Throwable) {
onError(
AudioException(
AudioException.Kind.DecoderError,
AudioException.Kind.EncoderError,
"Opus encode failed for a frame",
t,
),
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt
+17
View File
@@ -39,6 +39,14 @@ import com.vitorpamplona.quic.Varint
* `nestsClient/plans/2026-04-26-moq-lite-gap.md`.
*/
object MoqCodec {
/**
* Upper bound on a control-message payload length. MoQ-transport control
* messages are small (subscribes, announces, parameter blocks) — 16 MiB
* is well above any legitimate body and below any value where `.toInt()`
* would risk overflow. Anything bigger is treated as a malformed peer.
*/
const val MAX_CONTROL_PAYLOAD_LEN: Long = 16L * 1024L * 1024L
/**
* Encode a full control-stream frame (type + length + payload).
*/
@@ -65,6 +73,15 @@ object MoqCodec {
val typeDec = Varint.decode(src, offset) ?: return null
val lenDec = Varint.decode(src, offset + typeDec.bytesConsumed) ?: return null
val payloadStart = offset + typeDec.bytesConsumed + lenDec.bytesConsumed
// Bound the payload length before .toInt() — varint can carry up to
// 2^62, which silently overflows to a negative or absurd payloadEnd
// and makes the (payloadEnd > src.size) check return null forever,
// wedging the parser. Reject anything larger than the cap up front.
if (lenDec.value < 0L || lenDec.value > MAX_CONTROL_PAYLOAD_LEN) {
throw MoqCodecException(
"MoQ control payload length out of bounds: ${lenDec.value} (max $MAX_CONTROL_PAYLOAD_LEN)",
)
}
val payloadEnd = payloadStart + lenDec.value.toInt()
if (payloadEnd > src.size) return null
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+95 -130
View File
@@ -25,6 +25,7 @@ import com.vitorpamplona.nestsclient.moq.MoqWriter
import com.vitorpamplona.nestsclient.transport.WebTransportSession
import com.vitorpamplona.quic.Varint
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.cancelAndJoin
@@ -202,37 +203,78 @@ class MoqLiteSession internal constructor(
bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
bidi.write(MoqLiteCodec.encodeSubscribe(request))
// Single long-running collector pump for the bidi's response
// side. Reads the SubscribeResponse, then keeps collecting
// until the peer FINs (or scope is cancelled, or
// handle.unsubscribe() FINs our side and the relay echoes).
// The flow completion IS the moq-lite-03 signal that the
// publisher has disconnected mid-broadcast — Lite-03 has no
// explicit "publisher gone" message; bidi close is it.
// Without this watch, the frames Channel below would never
// close on remote disconnect, and any consumer collecting
// from the wrapper-level [MoqLiteSubscribeHandle.frames]
// flow would sit silent indefinitely after a publisher
// cycle even though the relay is happy to serve a fresh
// subscribe under the same broadcast suffix.
// Single long-running collector for the bidi's whole lifetime.
// The collector parses the SubscribeResponse inline, signals it
// via [responseDeferred], then continues draining the bidi
// until the peer FINs (publisher disconnect under moq-lite
// Lite-03) or the transport tears down. On collector exit,
// the frames channel is closed so the consumer's
// `frames.consumeAsFlow()` ends naturally.
//
// Why a single pump (vs separate response read + death
// watch): the underlying QUIC stream's `incoming` is
// backed by `Channel<ByteArray>.consumeAsFlow()` — which
// CANCELS the channel when the first collect ends. A second
// collect on a fresh `bidi.incoming()` Flow would see an
// already-cancelled channel and fire prematurely. Keeping
// one collect alive sidesteps that entirely.
// moq-lite's subscribe-response is a single size-prefixed
// message on the response side of the bidi. Read incoming
// chunks into a buffer until the buffer holds a full payload,
// then stop. We don't need a separate collector pump because
// post-Ok the bidi is idle (group data flows on its own uni
// streams).
val responseBuffer = MoqLiteFrameBuffer()
val responsePayload = readSubscribeResponseFromBidi(bidi.incoming(), responseBuffer, id)
when (val resp = MoqLiteCodec.decodeSubscribeResponse(responsePayload)) {
// Why one collector (vs separate response read + lifetime
// watch): the underlying QUIC stream's `incoming` is backed
// by `Channel<ByteArray>.consumeAsFlow()` which CANCELS the
// channel when the first collect ends. A second collect on a
// fresh `bidi.incoming()` Flow would see the already-cancelled
// channel and fire prematurely.
//
// Why we need a lifetime watch at all: pumpAnnounceWatch only
// closes frames when the relay actively sends Announce(Ended).
// A silent transport black-hole never reaches that path; without
// a per-subscription bidi watch, consumers would hang forever
// on dead UDP paths.
val frames = Channel<MoqLiteFrame>(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val responseDeferred = CompletableDeferred<MoqLiteCodec.SubscribeResponse>()
scope.launch {
val responseBuffer = MoqLiteFrameBuffer()
var responseParsed = false
try {
bidi.incoming().collect { chunk ->
if (!responseParsed) {
responseBuffer.push(chunk)
val typeCode = responseBuffer.readVarint() ?: return@collect
val body = responseBuffer.readSizePrefixed() ?: return@collect
val out = MoqWriter()
out.writeVarint(typeCode)
out.writeVarint(body.size.toLong())
out.writeBytes(body)
val resp = MoqLiteCodec.decodeSubscribeResponse(out.toByteArray())
responseDeferred.complete(resp)
responseParsed = true
}
// Post-response chunks are silently discarded —
// moq-lite leaves the bidi idle post-Ok. The signal
// we care about is the flow's natural completion.
}
} catch (ce: CancellationException) {
if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(ce)
throw ce
} catch (t: Throwable) {
if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(t)
}
if (!responseDeferred.isCompleted) {
responseDeferred.completeExceptionally(
MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id"),
)
}
// Close any frames channel registered for this id. The
// subscription may not be registered (response was Dropped
// or arrival raced cleanup) — in that case the snapshot
// returns null and we no-op.
val removed = state.withLock { subscriptionsBySubscribeId.remove(id) }
removed?.frames?.close()
}
val resp =
try {
responseDeferred.await()
} catch (t: Throwable) {
runCatching { bidi.finish() }
throw t
}
when (resp) {
is MoqLiteCodec.SubscribeResponse.Dropped -> {
runCatching { bidi.finish() }
throw MoqLiteSubscribeException(
"publisher rejected subscribe id=$id: errorCode=${resp.drop.errorCode} " +
"reason='${resp.drop.reasonPhrase}'",
@@ -240,7 +282,6 @@ class MoqLiteSession internal constructor(
}
is MoqLiteCodec.SubscribeResponse.Ok -> {
val frames = Channel<MoqLiteFrame>(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val sub =
ListenerSubscription(
id = id,
@@ -626,43 +667,6 @@ class MoqLiteSession internal constructor(
return uni
}
/**
* Read a size-prefixed payload from a bidi, seeded with whatever's
* already in [buffer] (the ControlType byte may have arrived with
* extra bytes). Used internally by the publisher inbound dispatch.
*/
private suspend fun readSizePrefixedFromBidiInto(
incoming: kotlinx.coroutines.flow.Flow<ByteArray>,
buffer: MoqLiteFrameBuffer,
): ByteArray {
// Try the buffer first — first chunk often contains the whole
// body since moq-lite messages are small and arrive as single
// QUIC sends.
buffer.readSizePrefixed()?.let { return it }
var done: ByteArray? = null
try {
incoming.collect { chunk ->
buffer.push(chunk)
buffer.readSizePrefixed()?.let {
done = it
throw EarlyExit
}
}
} catch (_: EarlyExit) {
// expected
} catch (ce: CancellationException) {
throw ce
}
return done
?: throw MoqCodecException("incoming bidi closed before a complete size-prefixed body arrived")
}
private object EarlyExit : RuntimeException() {
private fun readResolve(): Any = EarlyExit
override fun fillInStackTrace(): Throwable = this
}
suspend fun close() {
if (closed) return
closed = true
@@ -688,67 +692,6 @@ class MoqLiteSession internal constructor(
check(!closed) { "MoqLiteSession is closed" }
}
/**
* Read a moq-lite-03 SubscribeResponse off the bidi response side.
* The wire format is `[type_varint][body_size_varint][body_bytes]`
* — type lives OUTSIDE the size prefix (see
* `rs/moq-lite/src/lite/subscribe.rs::SubscribeResponse::encode`).
*
* Walks chunks into [buffer] until both the type discriminator and
* the size-prefixed body have arrived, then returns the contiguous
* `type+size+body` byte slab so [MoqLiteCodec.decodeSubscribeResponse]
* can parse it self-contained.
*
* Throws [MoqLiteSubscribeException] if the bidi closes before a
* full message arrives — that's the relay rejecting the subscribe
* with FIN instead of a SubscribeDrop reply.
*/
private suspend fun readSubscribeResponseFromBidi(
incoming: kotlinx.coroutines.flow.Flow<ByteArray>,
buffer: MoqLiteFrameBuffer,
id: Long,
): ByteArray {
// Snapshot the buffer's pos before each varint so we can roll
// back if not enough bytes have arrived yet — without this, a
// partial varint advances `pos` and the next chunk's bytes
// can't reconstitute it.
var typeCode: Long? = null
var body: ByteArray? = null
try {
// Some bytes may already be buffered (extra arrived with a
// prior message); try first without waiting for new chunks.
typeCode = buffer.readVarint()
if (typeCode != null) body = buffer.readSizePrefixed()
if (body != null) throw EarlyExit
incoming.collect { chunk ->
buffer.push(chunk)
if (typeCode == null) typeCode = buffer.readVarint()
if (typeCode != null && body == null) body = buffer.readSizePrefixed()
if (body != null) throw EarlyExit
}
} catch (_: EarlyExit) {
// expected
} catch (ce: CancellationException) {
throw ce
}
if (typeCode == null) {
throw MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id")
}
if (body == null) {
throw MoqLiteSubscribeException(
"subscribe stream FIN mid-body for id=$id (type=$typeCode)",
)
}
// Re-emit the contiguous `[type][size][body]` slab so
// [MoqLiteCodec.decodeSubscribeResponse] can parse it
// self-contained.
val out = MoqWriter()
out.writeVarint(typeCode!!)
out.writeVarint(body!!.size.toLong())
out.writeBytes(body!!)
return out.toByteArray()
}
private class ListenerSubscription(
val id: Long,
val request: MoqLiteSubscribe,
@@ -827,8 +770,30 @@ class MoqLiteSession internal constructor(
if (publisherClosed) return false
if (inboundSubs.isEmpty()) return false
val group = currentGroup ?: openNextGroupLocked().also { currentGroup = it }
val framed = Varint.encode(payload.size.toLong()) + payload
runCatching { group.uni.write(framed) }
// Single-allocation framing: write the varint length
// directly into a buffer sized for `varint + payload`,
// then copy the payload after it. The previous shape
// (`Varint.encode(...) + payload`) allocated twice per
// Opus frame — once for the varint, once for the `+`
// concatenation — at 50 fps × N speakers that's measurable
// young-gen pressure on the audio hot path.
val payloadSize = payload.size
val varintLen = Varint.size(payloadSize.toLong())
val framed = ByteArray(varintLen + payloadSize)
Varint.writeTo(payloadSize.toLong(), framed, 0)
payload.copyInto(framed, varintLen)
val writeResult = runCatching { group.uni.write(framed) }
if (writeResult.isFailure) {
// The uni stream errored (peer reset, transport closed,
// FIN'd by removeInboundSubscription). Drop the dead
// stream so the next send opens a fresh group instead of
// re-trying on a corpse, and surface the failure to the
// caller — the prior contract of "always true on
// not-muted" silently masked publisher disconnects.
runCatching { group.uni.finish() }
currentGroup = null
return false
}
}
return true
}
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt
+28
View File
@@ -385,6 +385,34 @@ class MoqLiteSessionTest {
session.close()
}
@Test
fun frames_flow_completes_when_peer_FINs_the_subscribe_bidi() =
runBlocking {
val (clientSide, serverSide) = FakeWebTransport.pair()
val session = MoqLiteSession.client(clientSide, pumpScope)
val peerBidi =
async {
val (bidi, _) = nextSubscribeBidi(serverSide)
bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
bidi
}
val handle = session.subscribe("speakerY", "audio/data")
val bidi = withTimeout(2_000) { peerBidi.await() }
// Peer FINs the subscribe bidi — moq-lite Lite-03's signal
// for "publisher gone, no more data ever". Without the
// per-subscription bidi-watch, this would leave the
// consumer-facing frames flow hanging forever; with the
// watch, the frames Channel closes and `toList()` returns.
bidi.finish()
withTimeout(2_000) { handle.frames.toList() }
session.close()
}
@Test
fun unsubscribe_FINs_the_subscribe_bidi() =
runBlocking {