nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(nestsclient): RESET_STREAM with typed code after SubscribeDrop body — moq-lite Lite-03

Browse Source 
moq-lite Lite-03 conveys application-level errors on any stream via
`RESET_STREAM(application_error_code u32)` (audit M3). Pre-fix, our
publisher's two SubscribeDrop reply paths
(`BROADCAST_DOES_NOT_EXIST`, `TRACK_DOES_NOT_EXIST`) wrote the Drop
body and then `bidi.finish()` — a graceful FIN — which overlapped
with "publisher gracefully shut down." A subscriber watching only
the QUIC layer (no body decode) couldn't tell rejection from
shutdown.

Replace the trailing `bidi.finish()` with `bidi.reset(errorCode)`
in both Drop paths. The errorCode matches the body's `errorCode`
field so a peer that decodes the Drop sees the same number as one
that only sees the RESET_STREAM frame. The plumbing landed in the
prior commit (`feat(quic,nestsclient): plumb RESET_STREAM +
STOP_SENDING through WebTransport`).

Tests: existing `publisher_replies_subscribeDrop_when_*` regressions
gain an additional `lastPeerResetCode` assertion via
`FakeBidiStream`, locking the typed-error contract.

Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (M3).

https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq
This commit is contained in:
Claude
2026-05-09 14:34:39 +00:00
parent 1c7b158e11
commit c7a49d1679
2 changed files with 47 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+20 -7
View File
@@ -933,7 +933,7 @@ class MoqLiteSession internal constructor(
if (sub.broadcast != ourSuffix) {
Log.w("NestTx") {
"SUBSCRIBE inbound id=${sub.id} broadcast='${sub.broadcast}' does not match " +
"publisher.suffix='$ourSuffix' — replying SubscribeDrop"
"publisher.suffix='$ourSuffix' — replying SubscribeDrop+RESET"
}
runCatching {
bidi.write(
@@ -946,7 +946,16 @@ class MoqLiteSession internal constructor(
),
),
)
bidi.finish()
// 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)
}
dispatched = true
return@collect
@@ -956,19 +965,23 @@ class MoqLiteSession internal constructor(
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
// error code BEFORE we RESET — without this the
// peer's response wait resolves only on
// bidi-FIN with no indication WHY (looks
// 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.
// 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"
"on this session (have ${publishersSnapshot.map { it.track }}) — replying SubscribeDrop+RESET"
}
runCatching {
bidi.write(
@@ -981,7 +994,7 @@ class MoqLiteSession internal constructor(
),
),
)
bidi.finish()
bidi.reset(MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST)
}
dispatched = true
return@collect
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt
+27
View File
@@ -672,6 +672,20 @@ class MoqLiteSessionTest {
kotlin.test.assertContains(dropped.drop.reasonPhrase, "wrongPubkey")
kotlin.test.assertContains(dropped.drop.reasonPhrase, "speakerPubkey")
// Audit M3: in addition to the application-level Drop body,
// the publisher must also `RESET_STREAM(errorCode)` the bidi
// so the peer can distinguish a typed rejection from a
// graceful "publisher gone" FIN. Drain so the read side
// closes and the cell publishes, then assert the code.
// toList() will end because reset() closed the channel.
withTimeout(2_000) { subBidi.incoming().toList() }
val fakeBidi = subBidi as FakeBidiStream
assertEquals(
MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST as Long?,
fakeBidi.lastPeerResetCode,
"publisher must reset the bidi with BROADCAST_DOES_NOT_EXIST after the Drop body",
)
publisher.close()
session.close()
}
@@ -716,6 +730,19 @@ class MoqLiteSessionTest {
// the exact text so we can keep tweaking the wording.
kotlin.test.assertContains(dropped.drop.reasonPhrase, "video/data")
// Audit M3: typed RESET_STREAM follows the Drop body. Same
// shape as broadcast_does_not_match — the code matches
// the Drop's errorCode so a peer that only watched the
// QUIC layer (no body decode) still sees the typed
// rejection.
withTimeout(2_000) { subBidi.incoming().toList() }
val fakeBidi = subBidi as FakeBidiStream
assertEquals(
MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST as Long?,
fakeBidi.lastPeerResetCode,
"publisher must reset the bidi with TRACK_DOES_NOT_EXIST after the Drop body",
)
publisher.close()
session.close()
}