From c7a49d1679e25a182645fac56e656a0f1d87156b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 9 May 2026 14:34:39 +0000 Subject: [PATCH] =?UTF-8?q?fix(nestsclient):=20RESET=5FSTREAM=20with=20typ?= =?UTF-8?q?ed=20code=20after=20SubscribeDrop=20body=20=E2=80=94=20moq-lite?= =?UTF-8?q?=20Lite-03?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../nestsclient/moq/lite/MoqLiteSession.kt | 27 ++++++++++++++----- .../moq/lite/MoqLiteSessionTest.kt | 27 +++++++++++++++++++ 2 files changed, 47 insertions(+), 7 deletions(-) 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 4b1d10814..82d247105 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 @@ -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 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 df333ffa7..00d6be8f5 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 @@ -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() }