From 5a86cdd4e41615832059c2722ec0beb0f92cc51d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 20:55:04 +0000 Subject: [PATCH] fix(audio-rooms): SubscribeResponse framing matches moq-lite-03 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lite-03's SubscribeResponse on the response side of a Subscribe bidi is two pieces concatenated: type varint (0 = Ok, 1 = Drop) body size-prefixed bytes The type discriminator sits OUTSIDE the body's size prefix. Earlier drafts wrapped the whole thing in one outer size prefix; Lite-03 split them. See `rs/moq-lite/src/lite/subscribe.rs::SubscribeResponse::encode` (the `_` arm, which matches Lite-03+). Our codec was producing — and reading — the older outer-wrapped form, so against a Lite-03 relay we mis-parsed the type discriminator (`0` for Ok) as a "size=0" outer prefix and surfaced a confusing `MoqCodecException: truncated varint at offset=0 (remaining=0)` from inside `decodeSubscribeResponse`. The first byte the relay sent was the type, not a size, and our reader peeled it off as the outer length. Fix touches three pieces: * `encodeSubscribeOk` / `encodeSubscribeDrop` emit `type + size_prefixed(body)` directly (no outer wrap), via a small `prefixWithType` helper. * `decodeSubscribeResponse` reads `type` then a length-prefixed body, decodes the body in its own reader, and asserts the outer payload is fully consumed. * `readSubscribeResponseFromBidi` walks chunks into the buffer until both the type varint and the size-prefixed body have arrived, then re-emits the contiguous `[type][size][body]` slab so the decoder parses it self-contained. Reuses the `EarlyExit` collector pattern the publisher-inbound path already uses; avoids `Flow.iterator()` which doesn't exist in kotlinx.coroutines. Tests: * `MoqLiteCodecTest::subscribe{Ok,Drop}_round_trips` no longer `peelSizePrefix` the encoded bytes — the new wire form has no outer prefix to strip; the bytes go straight through `decodeSubscribeResponse`. * `MoqLiteSessionTest::publisher_acks_subscribe_and_pushes_group_data_on_uni_stream` also drops `MoqLiteFrameBuffer().readSizePrefixed()` on the ack chunk for the same reason. Verified end-to-end against a bare-metal moq-relay 0.10.25 + moq-auth (`-DnestsInteropExternal=true`): * `NostrNestsRoundTripInteropTest::production_speaker_broadcasts_to_production_listener_via_real_relay` passes — speaker → listener → 8 frames round-trip cleanly. * `NostrNestsMultiPeerInteropTest::one_speaker_fans_out_to_two_listeners` passes — same speaker reaches both listeners with the full frame stream. * The `listener_subscribed_before_announce_receives_late_frames` and `two_speakers_in_same_room_deliver_independently_to_one_listener` cases still fail with `subscribe stream FIN before reply` — those are different relay-behavior issues (the v1 relay seems to FIN subscribes against unannounced broadcasts and against a second speaker on the same listener), unrelated to the codec. --- .../nestsclient/moq/lite/MoqLiteCodec.kt | 51 ++++++++++---- .../nestsclient/moq/lite/MoqLiteMessages.kt | 12 +++- .../nestsclient/moq/lite/MoqLiteSession.kt | 66 ++++++++++++++----- .../nestsclient/moq/lite/MoqLiteCodecTest.kt | 7 +- .../moq/lite/MoqLiteSessionTest.kt | 10 +-- 5 files changed, 108 insertions(+), 38 deletions(-) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt index 1b4379a8a..a21a5fe8b 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt @@ -126,21 +126,37 @@ object MoqLiteCodec { fun encodeSubscribeOk(msg: MoqLiteSubscribeOk): ByteArray { val body = MoqWriter() - body.writeVarint(MoqLiteSubscribeResponseType.Ok.code) body.writeByte(msg.priority) body.writeByte(if (msg.ordered) 1 else 0) body.writeVarint(msg.maxLatencyMillis) body.writeVarint(encodeOptionalGroup(msg.startGroup)) body.writeVarint(encodeOptionalGroup(msg.endGroup)) - return wrapSizePrefixed(body) + return prefixWithType(MoqLiteSubscribeResponseType.Ok.code, body) } fun encodeSubscribeDrop(msg: MoqLiteSubscribeDrop): ByteArray { val body = MoqWriter() - body.writeVarint(MoqLiteSubscribeResponseType.Drop.code) body.writeVarint(msg.errorCode) body.writeLengthPrefixedString(msg.reasonPhrase) - return wrapSizePrefixed(body) + return prefixWithType(MoqLiteSubscribeResponseType.Drop.code, body) + } + + /** + * moq-lite-03 SubscribeResponse framing: a top-level type + * discriminator varint, then a size-prefixed body. The type sits + * OUTSIDE the size prefix — see + * `rs/moq-lite/src/lite/subscribe.rs::SubscribeResponse::encode` + * (the `_` arm covers Lite03+). Earlier drafts size-prefixed the + * whole thing, but Lite03 uses this two-piece framing. + */ + private fun prefixWithType( + typeCode: Long, + body: MoqWriter, + ): ByteArray { + val out = MoqWriter() + out.writeVarint(typeCode) + out.writeBytes(wrapSizePrefixed(body)) + return out.toByteArray() } /** @@ -157,20 +173,29 @@ object MoqLiteCodec { ) : SubscribeResponse() } + /** + * Decode `[type_varint][body_size_varint][body]` as produced by + * [encodeSubscribeOk] / [encodeSubscribeDrop]. The body itself is + * size-prefixed; the type discriminator sits OUTSIDE that size + * prefix — see [prefixWithType] for the wire-format rationale. + */ fun decodeSubscribeResponse(payload: ByteArray): SubscribeResponse { val r = MoqReader(payload) val typeCode = r.readVarint() val type = MoqLiteSubscribeResponseType.fromCode(typeCode) ?: throw MoqCodecException("unknown moq-lite SubscribeResponse type: $typeCode") + val body = r.readLengthPrefixedBytes() + ensureFullyConsumed(r, "SubscribeResponse(type=$type)") + val br = MoqReader(body) return when (type) { MoqLiteSubscribeResponseType.Ok -> { - val priority = r.readByte() - val ordered = decodeOrderedByte(r.readByte()) - val maxLatencyMillis = r.readVarint() - val startGroup = decodeOptionalGroup(r.readVarint()) - val endGroup = decodeOptionalGroup(r.readVarint()) - ensureFullyConsumed(r, "SubscribeOk") + val priority = br.readByte() + val ordered = decodeOrderedByte(br.readByte()) + val maxLatencyMillis = br.readVarint() + val startGroup = decodeOptionalGroup(br.readVarint()) + val endGroup = decodeOptionalGroup(br.readVarint()) + ensureFullyConsumed(br, "SubscribeOk") SubscribeResponse.Ok( MoqLiteSubscribeOk( priority = priority, @@ -183,9 +208,9 @@ object MoqLiteCodec { } MoqLiteSubscribeResponseType.Drop -> { - val errorCode = r.readVarint() - val reason = r.readLengthPrefixedString() - ensureFullyConsumed(r, "SubscribeDrop") + val errorCode = br.readVarint() + val reason = br.readLengthPrefixedString() + ensureFullyConsumed(br, "SubscribeDrop") SubscribeResponse.Dropped( MoqLiteSubscribeDrop(errorCode = errorCode, reasonPhrase = reason), ) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt index d9b7f1af1..553274c94 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt @@ -101,10 +101,16 @@ enum class MoqLiteAnnounceStatus( } /** - * Type byte at the head of a [MoqLiteSubscribeResponse] payload. + * Type discriminator at the head of a SubscribeResponse on the response + * side of a Subscribe bidi. Wire layout (Lite-03+): * - * Source: `rs/moq-lite/src/lite/subscribe.rs:271-328`, - * `@moq/lite/lite/subscribe.js:264-282`. + * type varint (0 = Ok, 1 = Drop) + * body size-prefixed bytes + * + * The type sits OUTSIDE the body size prefix — see + * `rs/moq-lite/src/lite/subscribe.rs::SubscribeResponse::encode` (the + * `_` arm covers Lite03+). Earlier drafts wrapped type+body in one + * outer size prefix; Lite03 split them. */ enum class MoqLiteSubscribeResponseType( val code: Long, 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 de801f335..b61ac96cc 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 @@ -21,6 +21,7 @@ package com.vitorpamplona.nestsclient.moq.lite import com.vitorpamplona.nestsclient.moq.MoqCodecException +import com.vitorpamplona.nestsclient.moq.MoqWriter import com.vitorpamplona.nestsclient.transport.WebTransportSession import com.vitorpamplona.quic.Varint import kotlinx.coroutines.CancellationException @@ -514,29 +515,64 @@ class MoqLiteSession internal constructor( } /** - * Read the single size-prefixed subscribe response off a fresh - * subscribe bidi. moq-lite's subscribe-response is one wire write, - * so the first chunk we receive contains the entire payload (matches - * what the IETF [com.vitorpamplona.nestsclient.moq.MoqSession] does - * during its SETUP read). + * 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`). * - * Throws [MoqLiteSubscribeException] if the bidi closes before any - * chunk arrives. + * 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, buffer: MoqLiteFrameBuffer, id: Long, ): ByteArray { - val chunk = - incoming.firstOrNull() - ?: throw MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id") - buffer.push(chunk) - return buffer.readSizePrefixed() - ?: throw MoqLiteSubscribeException( - "first chunk on subscribe response did not carry a complete size-prefixed payload " + - "(id=$id, chunkSize=${chunk.size})", + // 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( diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodecTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodecTest.kt index ad5cb28ce..bbde1d16d 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodecTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodecTest.kt @@ -177,7 +177,10 @@ class MoqLiteCodecTest { startGroup = null, endGroup = null, ) - val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribeOk(msg)) + // moq-lite-03 SubscribeResponse is `[type][size][body]` with the + // type discriminator OUTSIDE the size prefix — the encoder no + // longer emits an outer wrap, so feed the bytes through directly. + val payload = MoqLiteCodec.encodeSubscribeOk(msg) val resp = MoqLiteCodec.decodeSubscribeResponse(payload) val ok = assertIs(resp) assertEquals(msg, ok.ok) @@ -186,7 +189,7 @@ class MoqLiteCodecTest { @Test fun subscribeDrop_round_trips() { val msg = MoqLiteSubscribeDrop(errorCode = 0x12L, reasonPhrase = "publisher gone") - val payload = peelSizePrefix(MoqLiteCodec.encodeSubscribeDrop(msg)) + val payload = MoqLiteCodec.encodeSubscribeDrop(msg) val resp = MoqLiteCodec.decodeSubscribeResponse(payload) val drop = assertIs(resp) assertEquals(msg, drop.drop) diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt index 09ff40cd0..36d8d6f82 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt @@ -300,12 +300,12 @@ class MoqLiteSessionTest { ), ) - // Step 2: we reply SubscribeOk. + // Step 2: we reply SubscribeOk. moq-lite-03's SubscribeResponse + // is `[type][size][body]` with the type discriminator OUTSIDE + // the size prefix — no outer wrap to strip, the chunk itself + // is the wire form decodeSubscribeResponse expects. val ackChunk = withTimeout(2_000) { subBidi.incoming().first() } - val resp = - MoqLiteCodec.decodeSubscribeResponse( - MoqLiteFrameBuffer().apply { push(ackChunk) }.readSizePrefixed()!!, - ) + val resp = MoqLiteCodec.decodeSubscribeResponse(ackChunk) assertIs(resp) // Step 3: publisher pushes one frame, which opens a uni