fix(audio-rooms): SubscribeResponse framing matches moq-lite-03

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.
2026-04-26 20:55:04 +00:00
parent da1c4d3968
commit 5a86cdd4e4
5 changed files with 108 additions and 38 deletions
@@ -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),
                )
@@ -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,
@@ -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<ByteArray>,
        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(
@@ -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<MoqLiteCodec.SubscribeResponse.Ok>(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<MoqLiteCodec.SubscribeResponse.Dropped>(resp)
        assertEquals(msg, drop.drop)
@@ -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<MoqLiteCodec.SubscribeResponse.Ok>(resp)

            // Step 3: publisher pushes one frame, which opens a uni