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 0ad9d08d6..59d8af06d 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 @@ -231,14 +231,40 @@ data class MoqLiteSubscribeOk( * RESET_STREAM, but the publisher can pre-emptively decline a * subscription with [MoqLiteSubscribeDrop] before any group flows. * - * Decode-only as far as the client cares — we never send Drop back - * upstream. + * Sent by [MoqLiteSession] when a relay opens a SUBSCRIBE bidi for a + * `track` we don't publish (e.g. a watcher subscribes to a `video/data` + * rendition but our broadcast only declares `audio/data` + `catalog.json` + * in its catalog). Without a Drop reply the watcher's response wait + * resolves only when the bidi is FIN'd, with no indication WHY — Drop + * carries the error code and a reason phrase the watcher can log / + * surface. + * + * Decoded by [MoqLiteSession] when the relay or upstream publisher + * rejects one of OUR subscriptions — we surface it as + * [MoqLiteSubscribeException] so callers see a typed protocol-level + * rejection rather than a silent end-of-flow. */ data class MoqLiteSubscribeDrop( val errorCode: Long, val reasonPhrase: String, ) +/** + * Error codes carried in the [MoqLiteSubscribeDrop.errorCode] varint. + * moq-lite leaves these application-defined; we mirror the IETF-MoQ + * conventions in `com.vitorpamplona.nestsclient.moq.ErrorCode` so a + * cross-protocol reader gets the same semantic from either path. + */ +object MoqLiteSubscribeDropCode { + /** + * The publisher does not serve this `(broadcast, track)` tuple. + * Sent for a subscribe whose `track` doesn't match any of the + * publishers we registered on this session. Mirrors + * `com.vitorpamplona.nestsclient.moq.ErrorCode.TRACK_DOES_NOT_EXIST`. + */ + const val TRACK_DOES_NOT_EXIST: Long = 0x04L +} + /** * Header at the start of a Group uni stream. After the * [MoqLiteDataType.Group] type byte, the publisher writes one 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 ff31ab993..bf85c60e4 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 @@ -849,11 +849,34 @@ 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 FIN — without this the + // peer's response wait resolves only on + // bidi-FIN 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. Log.w("NestTx") { "SUBSCRIBE inbound id=${sub.id} track='${sub.track}' has no matching publisher " + - "on this session (have ${publishersSnapshot.map { it.track }}) — finishing bidi" + "on this session (have ${publishersSnapshot.map { it.track }}) — replying SubscribeDrop" + } + 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.finish() } - runCatching { bidi.finish() } dispatched = true return@collect } 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 a7cad504d..70b1d5eb4 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 @@ -357,6 +357,50 @@ class MoqLiteSessionTest { session.close() } + @Test + fun publisher_replies_subscribeDrop_when_track_is_not_published() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + // Publisher serves audio/data only. + val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data") + + // Relay opens a Subscribe bidi for a DIFFERENT track. The + // session must reply with SubscribeDrop carrying the + // TRACK_DOES_NOT_EXIST code rather than a silent FIN — + // otherwise the watcher's response wait resolves only when + // the bidi is FIN'd, with no indication WHY (looks + // identical to "publisher disappeared mid-subscribe"). + val subBidi = serverSide.openBidiStream() + subBidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + subBidi.write( + MoqLiteCodec.encodeSubscribe( + MoqLiteSubscribe( + id = 99L, + broadcast = "speakerPubkey", + track = "video/data", + priority = 0x80, + ordered = true, + maxLatencyMillis = 0L, + startGroup = null, + endGroup = null, + ), + ), + ) + + val ackChunk = withTimeout(2_000) { subBidi.incoming().first() } + val resp = MoqLiteCodec.decodeSubscribeResponse(ackChunk) + val dropped = resp as MoqLiteCodec.SubscribeResponse.Dropped + assertEquals(MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST, dropped.drop.errorCode) + // Reason phrase is informational; pin substring rather than + // the exact text so we can keep tweaking the wording. + kotlin.test.assertContains(dropped.drop.reasonPhrase, "video/data") + + publisher.close() + session.close() + } + @Test fun publisher_close_emits_ended_announce() = runBlocking {