fix(nests): reply SubscribeDrop on unsupported tracks instead of silent FIN
When a relay opens a SUBSCRIBE bidi for a track this session doesn't publish (e.g. a watcher subscribes to `video/data` against a broadcast that only declares `audio/data` + `catalog.json`), the session was silently FINing the bidi without writing any reply. The peer's wait on its receive side resolves to end-of-stream with no indication WHY — indistinguishable from "publisher disappeared mid-subscribe." Reply with [MoqLiteSubscribeDrop] carrying [MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST] (0x04, mirroring the IETF-MoQ `ErrorCode.TRACK_DOES_NOT_EXIST` convention) and an informational reason phrase listing the tracks we DO publish, then FIN. Matches what kixelated's `rs/moq-lite/src/lite/subscribe.rs` expects on this path. Doesn't change the happy path — every track NestsUI-v2 subscribes to (`audio/data` + `catalog.json`) is now published — but is the spec-correct behaviour for any future watcher that prospectively subscribes to a track our publisher hasn't declared. https://claude.ai/code/session_014JfZJHSTvyYYWJbC9VbB47
This commit is contained in:
+28
-2
@@ -231,14 +231,40 @@ data class MoqLiteSubscribeOk(
|
|||||||
* RESET_STREAM, but the publisher can pre-emptively decline a
|
* RESET_STREAM, but the publisher can pre-emptively decline a
|
||||||
* subscription with [MoqLiteSubscribeDrop] before any group flows.
|
* subscription with [MoqLiteSubscribeDrop] before any group flows.
|
||||||
*
|
*
|
||||||
* Decode-only as far as the client cares — we never send Drop back
|
* Sent by [MoqLiteSession] when a relay opens a SUBSCRIBE bidi for a
|
||||||
* upstream.
|
* `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(
|
data class MoqLiteSubscribeDrop(
|
||||||
val errorCode: Long,
|
val errorCode: Long,
|
||||||
val reasonPhrase: String,
|
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
|
* Header at the start of a Group uni stream. After the
|
||||||
* [MoqLiteDataType.Group] type byte, the publisher writes one
|
* [MoqLiteDataType.Group] type byte, the publisher writes one
|
||||||
|
|||||||
+25
-2
@@ -849,11 +849,34 @@ class MoqLiteSession internal constructor(
|
|||||||
// single-track-per-publisher model, only one match is possible.
|
// single-track-per-publisher model, only one match is possible.
|
||||||
val targetPublisher = publishersSnapshot.firstOrNull { it.track == sub.track }
|
val targetPublisher = publishersSnapshot.firstOrNull { it.track == sub.track }
|
||||||
if (targetPublisher == null) {
|
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") {
|
Log.w("NestTx") {
|
||||||
"SUBSCRIBE inbound id=${sub.id} track='${sub.track}' has no matching publisher " +
|
"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
|
dispatched = true
|
||||||
return@collect
|
return@collect
|
||||||
}
|
}
|
||||||
|
|||||||
+44
@@ -357,6 +357,50 @@ class MoqLiteSessionTest {
|
|||||||
session.close()
|
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
|
@Test
|
||||||
fun publisher_close_emits_ended_announce() =
|
fun publisher_close_emits_ended_announce() =
|
||||||
runBlocking {
|
runBlocking {
|
||||||
|
|||||||
Reference in New Issue
Block a user