diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt index 3cf1cf575..3090ac990 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt @@ -109,20 +109,19 @@ suspend fun connectNestsListener( // moq-lite has NO setup message — the WebTransport handshake itself // is the handshake. Version (Lite-03 vs Lite-04) is selected by the - // ALPN exchange (`wt-available-protocols` / `wt-protocol`); fall - // back to Lite-03 if the server didn't echo `wt-protocol` (older - // single-protocol deployments). - val moqVersion = resolveMoqLiteVersion(webTransport.negotiatedSubProtocol) + // ALPN exchange (`wt-available-protocols` / `wt-protocol`) and + // surfaced via [MoqLiteSession.version]; the session itself derives + // it from `transport.negotiatedSubProtocol`. val moq = try { - MoqLiteSession.client(webTransport, scope, moqVersion) + MoqLiteSession.client(webTransport, scope) } catch (t: Throwable) { runCatching { webTransport.close(0, "moq-lite session init failed") } state.value = NestsListenerState.Failed("moq-lite session init failed: ${t.message}", t) return failedListener(state) } - state.value = NestsListenerState.Connected(room, versionCode(moqVersion)) + state.value = NestsListenerState.Connected(room, versionCode(moq.version)) return MoqLiteNestsListener( session = moq, mutableState = state, @@ -145,21 +144,6 @@ const val MOQ_LITE_03_VERSION: Long = 0x6D71_6C03L */ const val MOQ_LITE_04_VERSION: Long = 0x6D71_6C04L -/** - * Resolve the [MoqLiteVersion] to use for a session given the ALPN - * the WebTransport server selected (or `null` if the server didn't - * echo `wt-protocol`, e.g. older single-protocol deployments). - * - * Falls back to [MoqLiteVersion.LITE_03] when the negotiated value - * is unknown or absent — every nostrnests deployment supports - * Lite-03, so this stays safe under any of: - * - server is older than draft-13 (no `wt-protocol` echo) - * - server picked something we don't recognise (e.g. a future - * `moq-lite-05`) - * - client offered only one protocol - */ -internal fun resolveMoqLiteVersion(negotiatedSubProtocol: String?): MoqLiteVersion = MoqLiteVersion.fromAlpn(negotiatedSubProtocol) ?: MoqLiteVersion.LITE_03 - /** Map a [MoqLiteVersion] to its synthetic [MOQ_LITE_03_VERSION]-family code. */ internal fun versionCode(version: MoqLiteVersion): Long = when (version) { @@ -273,18 +257,18 @@ suspend fun connectNestsSpeaker( state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.MoqHandshake) // moq-lite has NO setup message. Same logic as the listener path — - // version is selected by the ALPN exchange. - val moqVersion = resolveMoqLiteVersion(webTransport.negotiatedSubProtocol) + // version is selected by the ALPN exchange and derived inside + // [MoqLiteSession.version]. val moq = try { - MoqLiteSession.client(webTransport, scope, moqVersion) + MoqLiteSession.client(webTransport, scope) } catch (t: Throwable) { runCatching { webTransport.close(0, "moq-lite session init failed") } state.value = NestsSpeakerState.Failed("moq-lite session init failed: ${t.message}", t) return failedSpeaker(state) } - state.value = NestsSpeakerState.Connected(room, versionCode(moqVersion)) + state.value = NestsSpeakerState.Connected(room, versionCode(moq.version)) return MoqLiteNestsSpeaker( session = moq, speakerPubkeyHex = speakerPubkeyHex, 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 d16d9aed6..be47b28ba 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 @@ -77,24 +77,24 @@ class MoqLiteSession internal constructor( */ internal val transport: WebTransportSession, private val scope: CoroutineScope, - /** - * Wire-format version this session speaks. Selected at - * WebTransport CONNECT time via the - * `wt-available-protocols` / `wt-protocol` exchange (see - * [com.vitorpamplona.nestsclient.transport.WebTransportSession.negotiatedSubProtocol]). - * Defaults to [MoqLiteVersion.LITE_03] for backward compat with - * call sites that don't pass an explicit version (most tests + - * the legacy single-version factory). Production - * `connectNestsListener` / `connectNestsSpeaker` pass the - * resolved version from the negotiated ALPN. - * - * The version selects between the version-conditional codec - * branches in [MoqLiteCodec] for `Announce.hops`, - * `AnnouncePlease.excludeHop`, and `Probe.rtt`. All other - * messages are wire-identical between the two versions. - */ - val version: MoqLiteVersion = MoqLiteVersion.LITE_03, ) { + /** + * Wire-format version this session speaks. Derived from the + * WebTransport ALPN the server selected via + * `wt-available-protocols` / `wt-protocol` + * ([WebTransportSession.negotiatedSubProtocol]); falls back to + * [MoqLiteVersion.LITE_03] when the server didn't echo + * `wt-protocol` or echoed something we don't recognise (older + * single-protocol deployments / forward-compat). + * + * Selects between the version-conditional codec branches in + * [MoqLiteCodec] for `Announce.hops`, `AnnouncePlease.excludeHop`, + * and `Probe.rtt`. All other messages are wire-identical between + * the two versions. Tests drive this via + * `FakeWebTransport.pair(negotiatedSubProtocol = …)`. + */ + val version: MoqLiteVersion = + MoqLiteVersion.fromAlpn(transport.negotiatedSubProtocol) ?: MoqLiteVersion.LITE_03 private val state = Mutex() private val subscriptionsBySubscribeId: MutableMap = HashMap() private var nextSubscribeId: Long = 0L @@ -883,6 +883,31 @@ class MoqLiteSession internal constructor( } } + /** + * Reject an inbound Subscribe with a typed error code. Writes the + * `SubscribeDrop` body so a watcher that decodes the application + * message sees the typed reason, then `RESET_STREAM(errorCode)`s + * the bidi so a watcher inspecting only the QUIC layer also sees + * the same code (audit M3 — Lite-03 conveys errors via + * RESET_STREAM, distinguishing "publisher rejected" from + * "publisher gracefully shut down" / FIN). Errors during the + * write/reset are swallowed: the bidi is already on its way out. + */ + private suspend fun rejectSubscribe( + bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream, + errorCode: Long, + reason: String, + ) { + runCatching { + bidi.write( + MoqLiteCodec.encodeSubscribeDrop( + MoqLiteSubscribeDrop(errorCode = errorCode, reasonPhrase = reason), + ), + ) + bidi.reset(errorCode) + } + } + private suspend fun handleInboundBidi(bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream) { // Single long-running collector for the bidi's full lifetime. // Pre-fix this dispatch was split into a `firstOrNull()` to @@ -1031,28 +1056,13 @@ class MoqLiteSession internal constructor( "SUBSCRIBE inbound id=${sub.id} broadcast='${sub.broadcast}' does not match " + "publisher.suffix='$ourSuffix' — replying SubscribeDrop+RESET" } - runCatching { - bidi.write( - MoqLiteCodec.encodeSubscribeDrop( - MoqLiteSubscribeDrop( - errorCode = MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST, - reasonPhrase = - "broadcast '${sub.broadcast}' is not published on this session " + - "(we publish '$ourSuffix')", - ), - ), - ) - // Lite-03 conveys errors on any stream via - // `RESET_STREAM(application_error_code)` (audit - // M3). The Drop body is the application-level - // signal; the reset is the QUIC-level signal - // that carries the same code, distinguishing - // "publisher rejected this subscribe" from - // "publisher gracefully shut down" (which - // would be a plain FIN). Pre-fix we FINed, - // overlapping the two semantics. - bidi.reset(MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST) - } + rejectSubscribe( + bidi = bidi, + errorCode = MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST, + reason = + "broadcast '${sub.broadcast}' is not published on this session " + + "(we publish '$ourSuffix')", + ) dispatched = true return@collect } @@ -1060,38 +1070,17 @@ class MoqLiteSession internal constructor( // single-track-per-publisher model, only one match is possible. val targetPublisher = publishersSnapshot.firstOrNull { it.track == sub.track } if (targetPublisher == null) { - // Reply SubscribeDrop with a TRACK_DOES_NOT_EXIST - // error code BEFORE we RESET — without this the - // peer's response wait resolves only on - // bidi tear-down with no indication WHY (looks - // identical to "publisher disappeared mid- - // subscribe"). Drop carries the error code + - // reason phrase the watcher can log / - // surface, and matches what kixelated's - // `rs/moq-lite/src/lite/subscribe.rs` - // expects for an unrecognised track on a - // live broadcast. RESET (audit M3) replaces - // the prior FIN: Lite-03 conveys errors via - // RESET_STREAM, distinguishing "publisher - // rejected" from "publisher gracefully shut - // down." Log.w("NestTx") { "SUBSCRIBE inbound id=${sub.id} track='${sub.track}' has no matching publisher " + "on this session (have ${publishersSnapshot.map { it.track }}) — replying SubscribeDrop+RESET" } - runCatching { - bidi.write( - MoqLiteCodec.encodeSubscribeDrop( - MoqLiteSubscribeDrop( - errorCode = MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST, - reasonPhrase = - "track '${sub.track}' is not published on this broadcast " + - "(available: ${publishersSnapshot.joinToString(",") { it.track }})", - ), - ), - ) - bidi.reset(MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST) - } + rejectSubscribe( + bidi = bidi, + errorCode = MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST, + reason = + "track '${sub.track}' is not published on this broadcast " + + "(available: ${publishersSnapshot.joinToString(",") { it.track }})", + ) dispatched = true return@collect } @@ -1286,8 +1275,8 @@ class MoqLiteSession internal constructor( // groups of any LOWER-priority track via the high byte. // Production sessions cycle on JWT refresh every 9 min, so the // wrap is defensive only. - val seq23 = sequence.coerceAtMost(0x7F_FFFFL).toInt() and 0x007F_FFFF - val packedPriority = ((trackPriority and 0xFF) shl 23) or seq23 + val seqLow = sequence.coerceAtMost(SEQ_MAX.toLong()).toInt() and SEQ_MASK + val packedPriority = ((trackPriority and 0xFF) shl SEQ_BITS) or seqLow uni.setPriority(packedPriority) uni.write(Varint.encode(MoqLiteDataType.Group.code)) uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence))) @@ -1607,6 +1596,19 @@ class MoqLiteSession internal constructor( */ const val DEFAULT_TRACK_PRIORITY: Int = 0x80 + /** + * Bit layout for the QUIC stream priority value packed by + * [openGroupStream]. Bit 31 is reserved as the sign bit (so + * the writer's signed `sortedByDescending` orders correctly); + * bits 30..23 carry [DEFAULT_TRACK_PRIORITY]-family + * `trackPriority u8`; bits 22..0 carry the low [SEQ_BITS] of + * the group sequence. See the kdoc on [openGroupStream] for + * the full rationale. + */ + private const val SEQ_BITS: Int = 23 + private const val SEQ_MASK: Int = 0x007F_FFFF + private const val SEQ_MAX: Int = SEQ_MASK + /** * Bitrate hint (bits/sec) we report on inbound moq-lite Probe * bidis as a publisher. Mirrors the upper-bound of an Opus @@ -1638,7 +1640,6 @@ class MoqLiteSession internal constructor( fun client( transport: WebTransportSession, pumpScope: CoroutineScope, - version: MoqLiteVersion = MoqLiteVersion.LITE_03, - ): MoqLiteSession = MoqLiteSession(transport, pumpScope, version) + ): MoqLiteSession = MoqLiteSession(transport, pumpScope) } } 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 046234f22..200e89ac0 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 @@ -352,8 +352,13 @@ class MoqLiteSessionTest { // client-side announce-watch flow surfaces them // unchanged. Pre-fix the codec was Lite-03-only and // would have read the count and discarded the IDs. - val (clientSide, serverSide) = FakeWebTransport.pair() - val session = MoqLiteSession.client(clientSide, pumpScope, MoqLiteVersion.LITE_04) + // Drive the negotiated sub-protocol via the fake so the + // session derives `version = LITE_04` itself (matching the + // production path, where `MoqLiteSession.version` reads + // through `transport.negotiatedSubProtocol`). + val (clientSide, serverSide) = + FakeWebTransport.pair(negotiatedSubProtocol = MoqLiteAlpn.LITE_04) + val session = MoqLiteSession.client(clientSide, pumpScope) assertEquals(MoqLiteVersion.LITE_04, session.version) val peer = diff --git a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt index 4402239a1..9af9955c8 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt @@ -470,14 +470,7 @@ private class StrippedWtReadStreamAdapter( override fun incoming(): Flow = stripped.data override suspend fun stopSending(errorCode: Long) { - // Demux unconditionally wires `stopSending` for every surfaced - // stream — the contract in `StrippedWtStream` is "non-null on - // streams with a read side", which is true here. The `?:` - // fallback exists only to defend against a future demux bug - // that forgets to wire it; Unit is a safe no-op for tests - // that mock a stripped stream without a real receiver. - val ss = stripped.stopSending ?: return - ss(errorCode) + stripped.stopSending(errorCode) } } @@ -522,8 +515,7 @@ private class StrippedWtBidiStreamAdapter( } override suspend fun stopSending(errorCode: Long) { - val ss = stripped.stopSending ?: return - ss(errorCode) + stripped.stopSending(errorCode) } /** diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt index d7412b893..345298a8f 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt @@ -78,10 +78,11 @@ class StrippedWtStream( /** * Send `STOP_SENDING(applicationErrorCode)` on the receive half * (RFC 9000 §3.5) — asks the peer to RESET its corresponding - * send side. Always non-null on streams surfaced here: every - * stripped stream has a read side. First call wins. + * send side. Non-null because every stripped stream surfaced + * here has a read side; the demux always wires it. First call + * wins. */ - val stopSending: (suspend (errorCode: Long) -> Unit)? = null, + val stopSending: suspend (errorCode: Long) -> Unit, ) /**